776 lines
28 KiB
Python
776 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import time
|
|
from email.message import Message
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
|
|
from backend.app.db.models import AuditEvent, JobRun
|
|
from backend.app.services import action_execution, deployment_broker
|
|
from backend.app.services.action_execution import (
|
|
approval_execution_supported,
|
|
mcp_execution_supported,
|
|
)
|
|
from backend.app.services.action_registry import ACTION_RISK, ACTIONS, action_default_target
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
CONTROLLER = ROOT / "ops" / "gitea-deploy-broker"
|
|
ACTION = "hub.migration.prepare_permissions"
|
|
TARGET = "tower/itworx-mcp-hub-permissions"
|
|
STATE_PATH = "/mnt/cache/appdata/gitea-deploy-runners/state/itworx-mcp-hub.json"
|
|
LEGACY_BACKUP_PATH = "/srv/example/appdata/itworx-mcp-hub/backups"
|
|
|
|
|
|
@pytest.fixture
|
|
def hub():
|
|
path = CONTROLLER / "hub_permission_preparation.py"
|
|
assert path.is_file(), "the protected controller permission helper is missing"
|
|
spec = importlib.util.spec_from_file_location("hub_permission_preparation_test", path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@pytest.fixture
|
|
def controller(monkeypatch):
|
|
monkeypatch.syspath_prepend(str(CONTROLLER))
|
|
spec = importlib.util.spec_from_file_location(
|
|
"hub_permission_controller_test", CONTROLLER / "status_server.py"
|
|
)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def permission_fixture(hub, tmp_path, monkeypatch):
|
|
if os.name != "posix" or not hasattr(os, "geteuid") or os.geteuid() != 0:
|
|
pytest.skip("permission ownership fixture requires Linux root")
|
|
state_dir = tmp_path / "cache" / "appdata" / "gitea-deploy-runners" / "state"
|
|
state = state_dir / "itworx-mcp-hub.json"
|
|
lock = state_dir.parent / "deploy.lock"
|
|
state_dir.mkdir(parents=True)
|
|
|
|
state_bytes = b'{"fixture":"hub state"}\n'.ljust(4096, b" ")
|
|
state.write_bytes(state_bytes)
|
|
lock.write_bytes(b"")
|
|
os.chown(state, 0, 0)
|
|
os.chmod(state, 0o644)
|
|
os.chown(lock, 0, 0)
|
|
os.chmod(lock, 0o644)
|
|
|
|
monkeypatch.setattr(hub, "STATE_FILE", str(state))
|
|
monkeypatch.setattr(hub, "LOCK_FILE", str(lock))
|
|
monkeypatch.setattr(hub, "LOCK_TIMEOUT_SECONDS", 0.25)
|
|
monkeypatch.setattr(hub, "LOCK_POLL_INTERVAL_SECONDS", 0.01)
|
|
digest = hashlib.sha256(state_bytes).hexdigest()
|
|
return state, lock, state_bytes, digest
|
|
|
|
|
|
def state_target(state: Path, *, mode: str = "0600", content: bytes | None = None) -> dict:
|
|
payload = content or b'{"fixture":"hub state"}\n'.ljust(4096, b" ")
|
|
return {
|
|
"name": "state_file",
|
|
"path": str(state),
|
|
"uid": 0,
|
|
"gid": 0,
|
|
"mode": mode,
|
|
"desired_mode": "0600",
|
|
"would_change": mode != "0600",
|
|
"acl_present": False,
|
|
"size": len(payload),
|
|
"sha256": hashlib.sha256(payload).hexdigest(),
|
|
}
|
|
|
|
|
|
def test_permission_action_is_high_risk_and_has_one_fixed_target():
|
|
assert ACTION in ACTIONS
|
|
assert ACTION_RISK[ACTION] == "high"
|
|
assert action_default_target(ACTION) == TARGET
|
|
assert approval_execution_supported(ACTION)
|
|
assert mcp_execution_supported(ACTION)
|
|
|
|
|
|
def test_broker_posts_only_the_bounded_reason_to_fixed_signed_path(monkeypatch):
|
|
calls = []
|
|
broker = deployment_broker.DeploymentBrokerClient()
|
|
monkeypatch.setattr(broker, "_request", lambda *args, **kwargs: calls.append((args, kwargs)) or {})
|
|
|
|
broker.hub_migration_prepare_permissions(reason="prepare fixed Hub permissions" * 40)
|
|
|
|
assert calls[0][0] == ("POST", "/v1/appops/hub-migration/prepare-permissions")
|
|
assert calls[0][1]["payload"] == {"reason": ("prepare fixed Hub permissions" * 40)[:1000]}
|
|
|
|
|
|
def test_governed_execution_blocks_wrong_target_before_job_or_broker(tmp_path, monkeypatch):
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'blocked.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
monkeypatch.setattr(
|
|
deployment_broker.DeploymentBrokerClient,
|
|
"hub_migration_prepare_permissions",
|
|
lambda *args, **kwargs: pytest.fail("wrong target must not call the controller"),
|
|
)
|
|
with Session(engine) as session:
|
|
result = action_execution.execute_governed_action(
|
|
session,
|
|
action=ACTION,
|
|
target="tower/other",
|
|
source="chatgpt-mcp",
|
|
risk="high",
|
|
reason="prepare fixed Hub permissions",
|
|
original_policy_decision="allowed",
|
|
)
|
|
assert result["policyDecision"] == "blocked"
|
|
assert session.exec(select(JobRun)).all() == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"status,success",
|
|
[("prepared", True), ("already_compliant", True), ("partial", False), ("blocked", False)],
|
|
)
|
|
def test_governed_execution_audits_single_state_file_outcome(tmp_path, monkeypatch, status, success):
|
|
engine = create_engine(f"sqlite:///{tmp_path / f'{status}.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
result_payload = {
|
|
"status": status,
|
|
"revision": "a" * 40,
|
|
"changedTargets": [STATE_PATH] if status in {"prepared", "partial"} else [],
|
|
"changedTargetDetails": [],
|
|
"uncertainTargets": [STATE_PATH] if status == "partial" else [],
|
|
}
|
|
if status == "partial":
|
|
result_payload["changedTargetDetails"] = [
|
|
{
|
|
"name": "state_file",
|
|
"path": STATE_PATH,
|
|
"observedMode": "0600",
|
|
"sameIdentityAtPath": True,
|
|
}
|
|
]
|
|
if status in {"prepared", "already_compliant"}:
|
|
result_payload["verifiedTargets"] = [
|
|
{
|
|
"name": "state_file",
|
|
"path": STATE_PATH,
|
|
"uid": 0,
|
|
"gid": 0,
|
|
"mode": "0600",
|
|
"aclPresent": False,
|
|
}
|
|
]
|
|
monkeypatch.setattr(
|
|
deployment_broker.DeploymentBrokerClient,
|
|
"hub_migration_prepare_permissions",
|
|
lambda self, **kwargs: result_payload,
|
|
)
|
|
with Session(engine) as session:
|
|
result = action_execution.execute_governed_action(
|
|
session,
|
|
action=ACTION,
|
|
target=TARGET,
|
|
source="chatgpt-mcp",
|
|
risk="high",
|
|
reason="prepare fixed Hub permissions",
|
|
original_policy_decision="allowed",
|
|
)
|
|
job = session.get(JobRun, result["jobId"])
|
|
audit = session.get(AuditEvent, result["auditId"])
|
|
assert result["success"] is success
|
|
assert job.status == ("succeeded" if success else "failed")
|
|
assert audit.result == ("success" if success else "failed")
|
|
assert audit.target == TARGET
|
|
assert audit.after_state["status"] == status
|
|
assert audit.after_state["changedTargets"] == result_payload["changedTargets"]
|
|
assert audit.after_state["changedTargetDetails"] == result_payload["changedTargetDetails"]
|
|
|
|
|
|
@pytest.mark.parametrize("legacy_status", ["prepared", "already_compliant", "partial"])
|
|
def test_governed_execution_rejects_legacy_two_target_reports_as_unknown(
|
|
tmp_path, monkeypatch, legacy_status
|
|
):
|
|
engine = create_engine(f"sqlite:///{tmp_path / f'legacy-{legacy_status}.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
result_payload = {
|
|
"status": legacy_status,
|
|
"revision": "a" * 40,
|
|
"changedTargets": [LEGACY_BACKUP_PATH] if legacy_status != "already_compliant" else [],
|
|
"changedTargetDetails": [],
|
|
"uncertainTargets": [],
|
|
"verifiedTargets": [
|
|
{
|
|
"name": "backup_directory",
|
|
"path": LEGACY_BACKUP_PATH,
|
|
"uid": 99,
|
|
"gid": 100,
|
|
"mode": "0700",
|
|
"aclPresent": False,
|
|
},
|
|
{
|
|
"name": "state_file",
|
|
"path": STATE_PATH,
|
|
"uid": 0,
|
|
"gid": 0,
|
|
"mode": "0600",
|
|
"aclPresent": False,
|
|
},
|
|
],
|
|
}
|
|
monkeypatch.setattr(
|
|
deployment_broker.DeploymentBrokerClient,
|
|
"hub_migration_prepare_permissions",
|
|
lambda self, **kwargs: result_payload,
|
|
)
|
|
with Session(engine) as session:
|
|
result = action_execution.execute_governed_action(
|
|
session,
|
|
action=ACTION,
|
|
target=TARGET,
|
|
source="chatgpt-mcp",
|
|
risk="high",
|
|
reason="prepare fixed Hub permissions",
|
|
original_policy_decision="allowed",
|
|
)
|
|
audit = session.get(AuditEvent, result["auditId"])
|
|
assert result["success"] is False
|
|
assert result["status"] == "outcome_unknown"
|
|
assert result["result"] == "outcome_unknown"
|
|
assert result["uncertainTargets"] == [STATE_PATH]
|
|
assert audit.result == "outcome_unknown"
|
|
assert audit.after_state["uncertainTargets"] == [STATE_PATH]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"failure", ["broker_exception", "unreachable", "malformed_success", "malformed_partial"]
|
|
)
|
|
def test_governed_execution_marks_unresolved_controller_results_unknown(
|
|
tmp_path, monkeypatch, failure
|
|
):
|
|
engine = create_engine(f"sqlite:///{tmp_path / f'{failure}.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
calls = []
|
|
|
|
def broker_call(self, **_kwargs):
|
|
calls.append("called")
|
|
if failure == "broker_exception":
|
|
raise RuntimeError("request ended without a controller result")
|
|
if failure == "unreachable":
|
|
return {"status": "unreachable"}
|
|
if failure == "malformed_partial":
|
|
return {
|
|
"status": "partial",
|
|
"revision": "a" * 40,
|
|
"changedTargets": [LEGACY_BACKUP_PATH],
|
|
"changedTargetDetails": [],
|
|
"uncertainTargets": [],
|
|
}
|
|
return {
|
|
"status": "prepared",
|
|
"revision": "a" * 40,
|
|
"changedTargets": [STATE_PATH],
|
|
"changedTargetDetails": [],
|
|
"uncertainTargets": [],
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
deployment_broker.DeploymentBrokerClient,
|
|
"hub_migration_prepare_permissions",
|
|
broker_call,
|
|
)
|
|
with Session(engine) as session:
|
|
result = action_execution.execute_governed_action(
|
|
session,
|
|
action=ACTION,
|
|
target=TARGET,
|
|
source="chatgpt-mcp",
|
|
risk="high",
|
|
reason="prepare fixed Hub permissions",
|
|
original_policy_decision="allowed",
|
|
)
|
|
job = session.get(JobRun, result["jobId"])
|
|
audit = session.get(AuditEvent, result["auditId"])
|
|
|
|
assert calls == ["called"]
|
|
assert result["success"] is False
|
|
assert result["status"] == "outcome_unknown"
|
|
assert result["result"] == "outcome_unknown"
|
|
assert result["uncertainTargets"] == [STATE_PATH]
|
|
assert job.status == "failed"
|
|
assert "outcome is unknown" in job.result_summary
|
|
assert audit.result == "outcome_unknown"
|
|
assert audit.after_state["status"] == "outcome_unknown"
|
|
assert audit.after_state["changedTargets"] == []
|
|
|
|
|
|
def test_preparation_changes_only_fixed_state_file_and_is_idempotent(hub, tmp_path, monkeypatch):
|
|
state, lock, state_bytes, digest = permission_fixture(hub, tmp_path, monkeypatch)
|
|
state_identity = (state.stat().st_dev, state.stat().st_ino)
|
|
original_open = hub._open_pinned_object
|
|
opened = []
|
|
changed_objects = []
|
|
|
|
# Keep a legacy backup path with an intentionally irrelevant mode. The
|
|
# helper must neither open nor chmod it.
|
|
legacy_backup = tmp_path / "user" / "appdata" / "itworx-mcp-hub" / "backups"
|
|
legacy_backup.mkdir(parents=True)
|
|
os.chmod(legacy_backup, 0o750)
|
|
monkeypatch.setattr(hub, "BACKUP_DIRECTORY", str(legacy_backup), raising=False)
|
|
|
|
def record_open(name, path):
|
|
opened.append((name, path))
|
|
return original_open(name, path)
|
|
|
|
real_set_mode = hub._set_mode
|
|
|
|
def record_change(fd, mode):
|
|
st = os.fstat(fd)
|
|
changed_objects.append((st.st_dev, st.st_ino))
|
|
real_set_mode(fd, mode)
|
|
|
|
monkeypatch.setattr(hub, "_open_pinned_object", record_open)
|
|
monkeypatch.setattr(hub, "_set_mode", record_change)
|
|
|
|
code, report = hub.prepare_fixed_permissions()
|
|
|
|
assert code == 0
|
|
assert report["status"] == "applied"
|
|
assert report["changed_targets"] == [str(state)]
|
|
assert [target["path"] for target in report["targets"]] == [str(state)]
|
|
assert stat.S_IMODE(state.stat().st_mode) == 0o600
|
|
assert stat.S_IMODE(lock.stat().st_mode) == 0o644
|
|
assert stat.S_IMODE(legacy_backup.stat().st_mode) == 0o750
|
|
assert state.read_bytes() == state_bytes
|
|
assert hashlib.sha256(state.read_bytes()).hexdigest() == digest
|
|
assert (state.stat().st_dev, state.stat().st_ino) == state_identity
|
|
assert (state.stat().st_uid, state.stat().st_gid) == (0, 0)
|
|
assert report["lock"]["acquired"] is True
|
|
assert opened == [("deploy_lock", str(lock)), ("state_file", str(state))]
|
|
assert changed_objects == [state_identity]
|
|
|
|
code, report = hub.prepare_fixed_permissions()
|
|
assert code == 0
|
|
assert report["status"] == "already_compliant"
|
|
assert report["changed_targets"] == []
|
|
assert [target["path"] for target in report["targets"]] == [str(state)]
|
|
assert stat.S_IMODE(legacy_backup.stat().st_mode) == 0o750
|
|
|
|
|
|
def test_dry_run_does_not_change_state(hub, tmp_path, monkeypatch):
|
|
state, lock, before, _digest = permission_fixture(hub, tmp_path, monkeypatch)
|
|
|
|
code, report = hub.execute()
|
|
|
|
assert code == 0
|
|
assert report["status"] == "ready"
|
|
assert [item["path"] for item in report["targets"]] == [str(state)]
|
|
assert stat.S_IMODE(state.stat().st_mode) == 0o644
|
|
assert stat.S_IMODE(lock.stat().st_mode) == 0o644
|
|
assert state.read_bytes() == before
|
|
|
|
|
|
def test_preparation_rechecks_fresh_state_digest_before_changing_mode(hub, tmp_path, monkeypatch):
|
|
state, _lock, _before, _digest = permission_fixture(hub, tmp_path, monkeypatch)
|
|
real_apply = hub._apply_changes
|
|
|
|
def change_before_apply(lock, pinned_state, mode, expected_digest):
|
|
state.write_bytes(b"changed after inspection".ljust(4096, b" "))
|
|
return real_apply(lock, pinned_state, mode, expected_digest)
|
|
|
|
monkeypatch.setattr(hub, "_apply_changes", change_before_apply)
|
|
|
|
code, report = hub.prepare_fixed_permissions()
|
|
|
|
assert code == 2
|
|
assert report["status"] == "failed"
|
|
assert "digest mismatch" in report["error"]
|
|
assert stat.S_IMODE(state.stat().st_mode) == 0o644
|
|
|
|
|
|
def test_symlinks_wrong_owner_mode_hardlinks_and_acl_fail_closed(hub, tmp_path, monkeypatch):
|
|
state, _lock, _before, _digest = permission_fixture(hub, tmp_path, monkeypatch)
|
|
|
|
real_state = state.with_name("state-real.json")
|
|
state.rename(real_state)
|
|
state.symlink_to(real_state)
|
|
code, report = hub.execute()
|
|
assert code == 2 and report["status"] == "blocked"
|
|
state.unlink()
|
|
real_state.rename(state)
|
|
|
|
os.chown(state, 99, 100)
|
|
code, report = hub.execute()
|
|
assert code == 2 and "owner mismatch" in report["error"]
|
|
os.chown(state, 0, 0)
|
|
|
|
os.link(state, state.with_name("unexpected-hardlink"))
|
|
code, report = hub.execute()
|
|
assert code == 2 and report["status"] == "blocked"
|
|
assert "link-count mismatch" in report["error"]
|
|
state.with_name("unexpected-hardlink").unlink()
|
|
|
|
os.chmod(state, 0o640)
|
|
code, report = hub.execute()
|
|
assert code == 2 and "mode mismatch" in report["error"]
|
|
os.chmod(state, 0o644)
|
|
|
|
state_identity = (state.stat().st_dev, state.stat().st_ino)
|
|
actual_has_acl = hub._has_acl
|
|
monkeypatch.setattr(
|
|
hub,
|
|
"_has_acl",
|
|
lambda fd: (os.fstat(fd).st_dev, os.fstat(fd).st_ino) == state_identity
|
|
or actual_has_acl(fd),
|
|
)
|
|
code, report = hub.execute()
|
|
assert code == 2 and "POSIX ACL is present" in report["error"]
|
|
assert stat.S_IMODE(state.stat().st_mode) == 0o644
|
|
|
|
|
|
def test_unsafe_deploy_lock_metadata_blocks_without_state_mutation(hub, tmp_path, monkeypatch):
|
|
state, lock, _before, _digest = permission_fixture(hub, tmp_path, monkeypatch)
|
|
os.chmod(lock, 0o600)
|
|
|
|
code, report = hub.prepare_fixed_permissions()
|
|
|
|
assert code == 2 and report["status"] == "blocked"
|
|
assert "mode mismatch" in report["error"]
|
|
assert stat.S_IMODE(state.stat().st_mode) == 0o644
|
|
assert stat.S_IMODE(lock.stat().st_mode) == 0o600
|
|
|
|
|
|
def test_replaced_state_path_reports_uncertain_outcome(hub, tmp_path, monkeypatch):
|
|
state, _lock, _before, _digest = permission_fixture(hub, tmp_path, monkeypatch)
|
|
moved_state = state.with_name("state-pinned.json")
|
|
real_set_mode = hub._set_mode
|
|
replaced = False
|
|
|
|
def replace_after_mode_change(fd, mode):
|
|
nonlocal replaced
|
|
real_set_mode(fd, mode)
|
|
if not replaced:
|
|
replaced = True
|
|
state.rename(moved_state)
|
|
state.write_bytes(b"replacement")
|
|
os.chown(state, 0, 0)
|
|
os.chmod(state, 0o644)
|
|
|
|
monkeypatch.setattr(hub, "_set_mode", replace_after_mode_change)
|
|
|
|
code, report = hub.prepare_fixed_permissions()
|
|
|
|
assert code == 2
|
|
assert report["status"] == "partial"
|
|
assert report["changed_targets"] == [
|
|
{
|
|
"name": "state_file",
|
|
"path": str(state),
|
|
"observed_mode": "0600",
|
|
"same_identity_at_path": False,
|
|
}
|
|
]
|
|
assert report["uncertain_targets"] == [str(state)]
|
|
assert stat.S_IMODE(moved_state.stat().st_mode) == 0o600
|
|
assert stat.S_IMODE(state.stat().st_mode) == 0o644
|
|
|
|
|
|
def test_fsync_failure_reports_changed_and_uncertain_state(hub, tmp_path, monkeypatch):
|
|
state, _lock, _before, _digest = permission_fixture(hub, tmp_path, monkeypatch)
|
|
monkeypatch.setattr(hub, "_fsync", lambda _fd: (_ for _ in ()).throw(OSError("fsync failed")))
|
|
|
|
code, report = hub.prepare_fixed_permissions()
|
|
|
|
assert code == 2
|
|
assert report["status"] == "partial"
|
|
assert report["changed_targets"] == [
|
|
{
|
|
"name": "state_file",
|
|
"path": str(state),
|
|
"observed_mode": "0600",
|
|
"same_identity_at_path": True,
|
|
}
|
|
]
|
|
assert report["uncertain_targets"] == [str(state)]
|
|
|
|
|
|
def test_bounded_deploy_lock_contention_fails_without_mutation(hub, tmp_path, monkeypatch):
|
|
state, lock, _before, _digest = permission_fixture(hub, tmp_path, monkeypatch)
|
|
fd = os.open(lock, os.O_RDONLY)
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
started = time.monotonic()
|
|
code, report = hub.prepare_fixed_permissions()
|
|
elapsed = time.monotonic() - started
|
|
finally:
|
|
fcntl.flock(fd, fcntl.LOCK_UN)
|
|
os.close(fd)
|
|
|
|
assert code == 2
|
|
assert report["status"] == "blocked"
|
|
assert "bounded wait" in report["error"]
|
|
assert elapsed < 1.0
|
|
assert stat.S_IMODE(state.stat().st_mode) == 0o644
|
|
|
|
|
|
def controller_handler(controller, monkeypatch, payload=None):
|
|
instance = object.__new__(controller.Handler)
|
|
body = json.dumps(payload or {"reason": "Prepare fixed Hub permissions"}).encode()
|
|
instance.headers = Message()
|
|
instance.headers["Content-Length"] = str(len(body))
|
|
instance.headers["Content-Type"] = "application/json"
|
|
instance.rfile = io.BytesIO(body)
|
|
instance.path = controller.HUB_PERMISSION_PREPARATION_PATH
|
|
responses = []
|
|
signatures = []
|
|
monkeypatch.setattr(instance, "_trusted_client", lambda: True)
|
|
monkeypatch.setattr(instance, "_send_json", lambda code, data: responses.append((code, json.loads(data))))
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_self_upgrade_signature_is_valid",
|
|
lambda *args, **kwargs: signatures.append(kwargs) or True,
|
|
)
|
|
monkeypatch.setattr(controller, "_self_upgrade_runtime", lambda: {"revision": "a" * 40})
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_require_controller_release",
|
|
lambda: {
|
|
"sourceRevision": "a" * 40,
|
|
"sourceRepository": controller.SELF_UPGRADE_REPOSITORY,
|
|
},
|
|
)
|
|
monkeypatch.setattr(controller, "_self_upgrade_preflight", lambda revision: {"status": "already_deployed"})
|
|
monkeypatch.setattr(controller, "_load_projects", lambda: {controller.SELF_UPGRADE_REPOSITORY: {"branch": "main"}})
|
|
monkeypatch.setattr(controller, "_repository_path_for_delivery", lambda *args: Path("fixed-repo.git"))
|
|
monkeypatch.setattr(controller, "_is_current_branch_head", lambda *args: True)
|
|
return instance, responses, signatures
|
|
|
|
|
|
def controller_success_report(state_path: str, *, already: bool = False) -> dict:
|
|
return {
|
|
"status": "already_compliant" if already else "applied",
|
|
"changed_targets": [] if already else [state_path],
|
|
"uncertain_targets": [],
|
|
"targets": [state_target(Path(state_path))],
|
|
}
|
|
|
|
|
|
def test_controller_route_uses_only_signed_protected_validated_current_main(
|
|
controller, monkeypatch
|
|
):
|
|
instance, responses, signatures = controller_handler(controller, monkeypatch)
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
controller.hub_permission_preparation,
|
|
"prepare_fixed_permissions",
|
|
lambda: calls.append("prepare")
|
|
or (0, controller_success_report(STATE_PATH)),
|
|
)
|
|
|
|
instance.do_POST()
|
|
|
|
assert signatures == [{"method": "POST", "path": controller.HUB_PERMISSION_PREPARATION_PATH}]
|
|
assert calls == ["prepare"]
|
|
assert responses == [
|
|
(
|
|
200,
|
|
{
|
|
"status": "prepared",
|
|
"revision": "a" * 40,
|
|
"changedTargets": [STATE_PATH],
|
|
"changedTargetDetails": [],
|
|
"uncertainTargets": [],
|
|
"verifiedTargets": [
|
|
{
|
|
"name": "state_file",
|
|
"path": STATE_PATH,
|
|
"uid": 0,
|
|
"gid": 0,
|
|
"mode": "0600",
|
|
"aclPresent": False,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
def test_controller_route_rejects_legacy_two_target_success_report(controller, monkeypatch):
|
|
instance, responses, _signatures = controller_handler(controller, monkeypatch)
|
|
legacy = controller_success_report(STATE_PATH)
|
|
legacy["changed_targets"] = [LEGACY_BACKUP_PATH, STATE_PATH]
|
|
legacy["targets"].insert(
|
|
0,
|
|
{
|
|
"name": "backup_directory",
|
|
"path": LEGACY_BACKUP_PATH,
|
|
"uid": 99,
|
|
"gid": 100,
|
|
"mode": "0700",
|
|
"desired_mode": "0700",
|
|
"would_change": False,
|
|
"acl_present": False,
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
controller.hub_permission_preparation,
|
|
"prepare_fixed_permissions",
|
|
lambda: (0, legacy),
|
|
)
|
|
|
|
instance.do_POST()
|
|
|
|
assert responses == [
|
|
(
|
|
200,
|
|
{
|
|
"status": "outcome_unknown",
|
|
"revision": "a" * 40,
|
|
"changedTargets": [],
|
|
"changedTargetDetails": [],
|
|
"uncertainTargets": [STATE_PATH],
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
def test_controller_route_rejects_success_without_full_helper_projection(controller, monkeypatch):
|
|
instance, responses, _signatures = controller_handler(controller, monkeypatch)
|
|
malformed = controller_success_report(STATE_PATH, already=True)
|
|
malformed.pop("changed_targets")
|
|
monkeypatch.setattr(
|
|
controller.hub_permission_preparation,
|
|
"prepare_fixed_permissions",
|
|
lambda: (0, malformed),
|
|
)
|
|
|
|
instance.do_POST()
|
|
|
|
assert responses == [
|
|
(
|
|
200,
|
|
{
|
|
"status": "outcome_unknown",
|
|
"revision": "a" * 40,
|
|
"changedTargets": [],
|
|
"changedTargetDetails": [],
|
|
"uncertainTargets": [STATE_PATH],
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{"reason": "Prepare fixed Hub permissions", "path": "/"},
|
|
{"reason": "Prepare fixed Hub permissions", "mode": "0777"},
|
|
{"reason": "Prepare fixed Hub permissions", "command": "chmod -R 700 /"},
|
|
],
|
|
)
|
|
def test_controller_route_rejects_caller_selected_effects(controller, monkeypatch, payload):
|
|
instance, responses, _signatures = controller_handler(controller, monkeypatch, payload)
|
|
monkeypatch.setattr(
|
|
controller.hub_permission_preparation,
|
|
"prepare_fixed_permissions",
|
|
lambda: pytest.fail("invalid request must not inspect or mutate permissions"),
|
|
)
|
|
|
|
instance.do_POST()
|
|
|
|
assert responses == [(400, {"status": "invalid_request"})]
|
|
|
|
|
|
def test_controller_route_does_not_mutate_before_full_push_validation(controller, monkeypatch):
|
|
instance, responses, _signatures = controller_handler(controller, monkeypatch)
|
|
monkeypatch.setattr(controller, "_self_upgrade_preflight", lambda _revision: {"status": "preflight_rejected"})
|
|
monkeypatch.setattr(
|
|
controller.hub_permission_preparation,
|
|
"prepare_fixed_permissions",
|
|
lambda: pytest.fail("pending validation must block permission changes"),
|
|
)
|
|
|
|
instance.do_POST()
|
|
|
|
assert responses == [(409, {"status": "preflight_rejected", "revision": "a" * 40})]
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["helper_exception", "malformed_report", "malformed_partial"])
|
|
def test_controller_route_reports_unknown_state_after_unresolved_helper_call(
|
|
controller, monkeypatch, failure
|
|
):
|
|
instance, responses, _signatures = controller_handler(controller, monkeypatch)
|
|
if failure == "helper_exception":
|
|
def helper_failure():
|
|
raise RuntimeError("helper returned no result")
|
|
|
|
monkeypatch.setattr(
|
|
controller.hub_permission_preparation, "prepare_fixed_permissions", helper_failure
|
|
)
|
|
elif failure == "malformed_report":
|
|
monkeypatch.setattr(
|
|
controller.hub_permission_preparation,
|
|
"prepare_fixed_permissions",
|
|
lambda: (0, {"status": "applied", "changed_targets": "unexpected"}),
|
|
)
|
|
else:
|
|
monkeypatch.setattr(
|
|
controller.hub_permission_preparation,
|
|
"prepare_fixed_permissions",
|
|
lambda: (
|
|
2,
|
|
{
|
|
"status": "partial",
|
|
"changed_targets": [LEGACY_BACKUP_PATH],
|
|
"uncertain_targets": [],
|
|
},
|
|
),
|
|
)
|
|
|
|
instance.do_POST()
|
|
|
|
assert responses == [
|
|
(
|
|
200,
|
|
{
|
|
"status": "outcome_unknown",
|
|
"revision": "a" * 40,
|
|
"changedTargets": [],
|
|
"changedTargetDetails": [],
|
|
"uncertainTargets": [STATE_PATH],
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
def test_controller_artifact_manifests_include_new_fixed_helper(controller):
|
|
renderer_spec = importlib.util.spec_from_file_location(
|
|
"hub_permission_artifact_renderer", ROOT / "scripts/prepare_controller_artifact.py"
|
|
)
|
|
assert renderer_spec is not None and renderer_spec.loader is not None
|
|
renderer = importlib.util.module_from_spec(renderer_spec)
|
|
renderer_spec.loader.exec_module(renderer)
|
|
volume_spec = importlib.util.spec_from_file_location(
|
|
"hub_permission_volume_preparer", ROOT / "scripts/prepare_controller_hub_release_volume.py"
|
|
)
|
|
assert volume_spec is not None and volume_spec.loader is not None
|
|
volume = importlib.util.module_from_spec(volume_spec)
|
|
volume_spec.loader.exec_module(volume)
|
|
installer = (ROOT / "scripts/upgrade_deployment_controller.sh").read_text(encoding="utf-8")
|
|
|
|
assert "hub_permission_preparation.py" in controller._CONTROLLER_ARTIFACT_FILES
|
|
assert "hub_permission_preparation.py" in renderer._ARTIFACT_FILES
|
|
assert tuple(volume.RELEASE_FILES) == tuple(renderer._ARTIFACT_FILES)
|
|
assert "ops/gitea-deploy-broker/hub_permission_preparation.py" in installer
|
|
assert "hub_permission_preparation.py" in installer
|