37ed8b012b
The account task set groups: [] with append: false, so the next run would have removed custodian-agt from gated-executor's exchange group. Allowed groups are now an explicit per-account inventory list, refused if forbidden, and asserted as the exact set by goss; container test covers an unlisted group. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 605711@bnt-lap001 Assistant-Session: 1f5e2256-181f-49ed-b7b1-a1c34bf37a61
205 lines
9.4 KiB
Python
205 lines
9.4 KiB
Python
"""Offline contract for the restricted agent account (CUST-WP-0073-T02, RAIL-HO-WP-0014).
|
|
|
|
Static checks always run. The container check renders the role's sshd drop-in
|
|
and Goss spec and validates them against a real OpenSSH 9.6 in a throwaway
|
|
ubuntu:24.04 container; enable it with RAILIANCE_CONTAINER_TESTS=1. Nothing
|
|
here contacts a managed host.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import jinja2
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ROLE = ROOT / "ansible" / "roles" / "ssh_ca_host"
|
|
ACCOUNT = "custodian-agt"
|
|
PRINCIPAL = "agt-custodian"
|
|
FORBIDDEN_SSHD = {
|
|
"AllowAgentForwarding": "no",
|
|
"AllowTcpForwarding": "no",
|
|
"X11Forwarding": "no",
|
|
"PermitTunnel": "no",
|
|
"PermitUserRC": "no",
|
|
"DisableForwarding": "yes",
|
|
"AuthorizedKeysFile": "none",
|
|
"AuthenticationMethods": "publickey",
|
|
"PasswordAuthentication": "no",
|
|
"KbdInteractiveAuthentication": "no",
|
|
}
|
|
REQUIRED_FORBIDDEN_GROUPS = {"sudo", "adm", "docker", "systemd-journal", "root"}
|
|
|
|
|
|
def _load(path: Path):
|
|
return yaml.safe_load(path.read_text())
|
|
|
|
|
|
def render(template: Path, **ctx) -> str:
|
|
env = jinja2.Environment(
|
|
loader=jinja2.FileSystemLoader(str(ROOT)),
|
|
keep_trailing_newline=True,
|
|
trim_blocks=True,
|
|
undefined=jinja2.StrictUndefined,
|
|
)
|
|
env.filters["dict2items"] = lambda d: [{"key": k, "value": v} for k, v in d.items()]
|
|
return env.get_template(str(template.relative_to(ROOT))).render(**ctx)
|
|
|
|
|
|
def railiance01_context() -> dict:
|
|
principals = _load(ROOT / "ansible" / "inventory" / "ssh_principals.yaml")["ssh_principals"]
|
|
host = principals["Railiance01"]
|
|
ctx = dict(_load(ROLE / "defaults" / "main.yml"))
|
|
ctx["ssh_ca_host_restricted_accounts"] = host["restricted_accounts"]
|
|
ctx["ssh_ca_host_principals"] = host["users"]
|
|
ctx["ssh_ca_host_restricted_groups"] = host.get("restricted_account_groups", {})
|
|
return ctx
|
|
|
|
|
|
class RestrictedAgentInventoryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.principals = _load(ROOT / "ansible" / "inventory" / "ssh_principals.yaml")["ssh_principals"]
|
|
|
|
def test_principal_maps_only_to_restricted_account_on_railiance01(self) -> None:
|
|
holders = [
|
|
(host, user)
|
|
for host, data in self.principals.items()
|
|
for user, plist in (data.get("users") or {}).items()
|
|
if PRINCIPAL in plist
|
|
]
|
|
self.assertEqual(holders, [("Railiance01", ACCOUNT)])
|
|
self.assertEqual(self.principals["Railiance01"]["users"][ACCOUNT], [PRINCIPAL])
|
|
|
|
def test_restricted_accounts_are_declared_and_mapped(self) -> None:
|
|
for host, data in self.principals.items():
|
|
for account in data.get("restricted_accounts") or []:
|
|
self.assertIn(account, data["users"], f"{host}:{account} lacks principals")
|
|
self.assertNotIn(account, {"root", "tegwick"})
|
|
self.assertNotIn("restricted_accounts", self.principals["CoulombCore"])
|
|
|
|
def test_existing_operator_mappings_unchanged(self) -> None:
|
|
# Retirement of the generic agt-* principals from tegwick is a tracked
|
|
# follow-up (RAIL-HO-WP-0014-T03), not part of this change.
|
|
for host in ("CoulombCore", "Railiance01"):
|
|
self.assertEqual(
|
|
self.principals[host]["users"]["tegwick"],
|
|
["agt-task-bridge", "agt-interhub-bootstrap", "adm-full", "atm-backup-daily"],
|
|
)
|
|
|
|
|
|
class RestrictedAgentRoleTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tasks = _load(ROLE / "tasks" / "main.yml")
|
|
self.defaults = _load(ROLE / "defaults" / "main.yml")
|
|
|
|
def _task(self, name: str) -> dict:
|
|
matches = [t for t in self.tasks if t["name"] == name]
|
|
self.assertEqual(len(matches), 1, name)
|
|
return matches[0]
|
|
|
|
def test_account_has_locked_password_and_only_allowed_groups(self) -> None:
|
|
user = self._task("Ensure restricted agent account (locked password, only allowed supplementary groups)")[
|
|
"ansible.builtin.user"
|
|
]
|
|
# Groups come only from the explicit per-account inventory list.
|
|
self.assertEqual(user["groups"], "{{ ssh_ca_host_restricted_groups.get(item, []) }}")
|
|
self.assertIs(user["append"], False)
|
|
self.assertEqual(user["password"], "!")
|
|
self.assertEqual(user["update_password"], "always")
|
|
|
|
def test_allowed_groups_are_explicit_and_not_forbidden(self) -> None:
|
|
groups = railiance01_context()["ssh_ca_host_restricted_groups"]
|
|
self.assertEqual(groups, {ACCOUNT: ["gexec-review"]})
|
|
for allowed in groups[ACCOUNT]:
|
|
self.assertNotIn(allowed, REQUIRED_FORBIDDEN_GROUPS)
|
|
|
|
def test_home_0700_no_sudoers_no_authorized_keys(self) -> None:
|
|
home = self._task("Restrict restricted agent home to the account (0700)")["ansible.builtin.file"]
|
|
self.assertEqual(home["mode"], "0700")
|
|
sudoers = self._task("Ensure no sudoers drop-in exists for restricted agent accounts")[
|
|
"ansible.builtin.file"
|
|
]
|
|
self.assertEqual(sudoers["state"], "absent")
|
|
keys = self._task("Remove any authorized_keys for restricted agent accounts")
|
|
self.assertEqual(keys["ansible.builtin.file"]["state"], "absent")
|
|
self.assertIn("authorized_keys", keys["loop"])
|
|
|
|
def test_sshd_dropin_is_validated_and_sorts_before_hardening(self) -> None:
|
|
tmpl = self._task("Configure sshd Match blocks for restricted agent accounts")["ansible.builtin.template"]
|
|
self.assertEqual(tmpl["validate"], "/usr/sbin/sshd -t -f %s")
|
|
self.assertLess(Path(tmpl["dest"]).name, "10-hardening.conf")
|
|
|
|
def test_every_restricted_task_is_reachable_by_tag(self) -> None:
|
|
for task in self.tasks:
|
|
if "restricted" in task["name"].lower() or task["name"].startswith("Deploy auth_principals"):
|
|
self.assertIn("restricted_agent_account", task.get("tags", []), task["name"])
|
|
|
|
def test_forbidden_groups_and_unreadable_kubeconfig_declared(self) -> None:
|
|
self.assertTrue(REQUIRED_FORBIDDEN_GROUPS <= set(self.defaults["ssh_ca_host_restricted_agent_forbidden_groups"]))
|
|
self.assertIn("/etc/rancher/k3s/k3s.yaml", self.defaults["ssh_ca_host_restricted_agent_unreadable_paths"])
|
|
|
|
|
|
class RestrictedAgentRenderTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.ctx = railiance01_context()
|
|
|
|
def test_match_block_turns_forbidden_options_off(self) -> None:
|
|
text = render(ROLE / "templates" / "restricted-agents-sshd.conf.j2", **self.ctx)
|
|
lines = [line.strip() for line in text.splitlines() if line.strip() and not line.startswith("#")]
|
|
self.assertEqual(lines[0], f"Match User {ACCOUNT}")
|
|
options = dict(line.split(None, 1) for line in lines[1:])
|
|
for key, value in FORBIDDEN_SSHD.items():
|
|
self.assertEqual(options.get(key), value, key)
|
|
self.assertNotIn("PermitUserEnvironment", options) # global-only keyword
|
|
|
|
def test_goss_spec_asserts_required_posture(self) -> None:
|
|
spec = yaml.safe_load(render(ROOT / "goss" / "restricted-agent.yaml.j2", **self.ctx))
|
|
self.assertFalse(spec["file"][f"/etc/sudoers.d/{ACCOUNT}"]["exists"])
|
|
self.assertFalse(spec["file"][f"/home/{ACCOUNT}/.ssh/authorized_keys"]["exists"])
|
|
self.assertEqual(spec["file"][f"/home/{ACCOUNT}"]["mode"], "0700")
|
|
commands = spec["command"]
|
|
self.assertIn(f"sudo -n -l -U {ACCOUNT}", commands)
|
|
idg = commands[f"id -nG {ACCOUNT}"]["stdout"]
|
|
# Exact set: the account's own group plus gexec-review, nothing else.
|
|
self.assertIn(f"/^{ACCOUNT} gexec-review$/", idg)
|
|
for group in REQUIRED_FORBIDDEN_GROUPS:
|
|
self.assertIn(f"!/\\b{group}\\b/", idg)
|
|
self.assertTrue(any("k3s.yaml" in c and "runuser" in c for c in commands))
|
|
sshd = next(v for c, v in commands.items() if c.startswith(f"/usr/sbin/sshd -T -C user={ACCOUNT},"))
|
|
for key, value in FORBIDDEN_SSHD.items():
|
|
self.assertIn(f"/^{key.lower()} {value.lower()}$/", sshd["stdout"])
|
|
self.assertIn("/^permituserenvironment no$/", sshd["stdout"])
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("RAILIANCE_CONTAINER_TESTS") == "1", "set RAILIANCE_CONTAINER_TESTS=1")
|
|
class RestrictedAgentContainerTests(unittest.TestCase):
|
|
def test_real_sshd_and_goss_in_throwaway_container(self) -> None:
|
|
docker = shutil.which("docker")
|
|
if docker is None:
|
|
self.skipTest("docker not available")
|
|
ctx = railiance01_context()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
Path(tmp, "05-restricted-agents.conf").write_text(
|
|
render(ROLE / "templates" / "restricted-agents-sshd.conf.j2", **ctx)
|
|
)
|
|
Path(tmp, "restricted-agent.yaml").write_text(render(ROOT / "goss" / "restricted-agent.yaml.j2", **ctx))
|
|
shutil.copy(ROOT / "tests" / "fixtures" / "restricted_agent_container.sh", tmp)
|
|
os.chmod(tmp, 0o755)
|
|
result = subprocess.run(
|
|
[docker, "run", "--rm", "-v", f"{tmp}:/in:ro", "ubuntu:24.04", "bash", "/in/restricted_agent_container.sh"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=600,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stdout[-3000:] + result.stderr[-2000:])
|
|
self.assertIn("ALL CONTAINER CHECKS PASSED", result.stdout)
|
|
print(result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|