66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import copy
|
||
|
|
import sys
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
||
|
|
|
||
|
|
from handoff_contract import ( # noqa: E402
|
||
|
|
HandoffContractError,
|
||
|
|
validate_goss_commands,
|
||
|
|
validate_payload,
|
||
|
|
validate_playbook,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class ReadOnlyHandoffContractTests(unittest.TestCase):
|
||
|
|
@classmethod
|
||
|
|
def setUpClass(cls) -> None:
|
||
|
|
cls.path = ROOT / "ansible" / "playbooks" / "verify.yaml"
|
||
|
|
cls.payload = yaml.safe_load(cls.path.read_text(encoding="utf-8"))
|
||
|
|
|
||
|
|
def test_repository_playbook_is_read_only(self) -> None:
|
||
|
|
validate_playbook(self.path)
|
||
|
|
|
||
|
|
def test_remote_template_module_is_rejected(self) -> None:
|
||
|
|
payload = copy.deepcopy(self.payload)
|
||
|
|
payload[0]["tasks"].append(
|
||
|
|
{
|
||
|
|
"name": "Mutate remote config",
|
||
|
|
"ansible.builtin.template": {"src": "x", "dest": "/etc/goss/baseline.yaml"},
|
||
|
|
}
|
||
|
|
)
|
||
|
|
with self.assertRaisesRegex(HandoffContractError, "not read-only"):
|
||
|
|
validate_payload(payload)
|
||
|
|
|
||
|
|
def test_arbitrary_remote_command_is_rejected(self) -> None:
|
||
|
|
payload = copy.deepcopy(self.payload)
|
||
|
|
command = next(
|
||
|
|
task for task in payload[0]["tasks"] if "ansible.builtin.command" in task
|
||
|
|
)
|
||
|
|
command["ansible.builtin.command"]["argv"] = ["systemctl", "restart", "ssh"]
|
||
|
|
with self.assertRaisesRegex(HandoffContractError, "fixed Goss"):
|
||
|
|
validate_payload(payload)
|
||
|
|
|
||
|
|
def test_controller_write_requires_local_delegation(self) -> None:
|
||
|
|
payload = copy.deepcopy(self.payload)
|
||
|
|
output = next(task for task in payload[0]["tasks"] if "ansible.builtin.copy" in task)
|
||
|
|
del output["delegate_to"]
|
||
|
|
with self.assertRaisesRegex(HandoffContractError, "not read-only"):
|
||
|
|
validate_payload(payload)
|
||
|
|
|
||
|
|
def test_goss_command_surface_is_pinned(self) -> None:
|
||
|
|
validate_goss_commands(
|
||
|
|
ROOT / "goss" / "baseline.yaml.j2",
|
||
|
|
ROOT / "spec" / "server-baseline.yaml",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|