Remove 'enroll diff --enforce' option. Tighten yaml data re: handlers - use listen: instead of notify:
This commit is contained in:
parent
e9d7d74445
commit
903125976d
GPG key ID:
03906B4110AAD3B8
15 changed files with 851 additions and 1288 deletions
|
|
@ -6,11 +6,9 @@ from pathlib import Path
|
|||
|
||||
import pytest
|
||||
|
||||
from enroll.ansible import _role_tag
|
||||
from enroll.diff import (
|
||||
_Spinner,
|
||||
_enforcement_plan,
|
||||
has_enforceable_drift,
|
||||
_role_tag,
|
||||
_utc_now_iso,
|
||||
_report_markdown,
|
||||
)
|
||||
|
|
@ -798,150 +796,6 @@ def test_role_tag_empty():
|
|||
assert _role_tag(" ") == "role_other"
|
||||
|
||||
|
||||
def test_has_enforceable_drift_packages_removed():
|
||||
report = {"packages": {"removed": ["vim"]}}
|
||||
assert has_enforceable_drift(report) is True
|
||||
|
||||
|
||||
def test_has_enforceable_drift_services_removed():
|
||||
report = {"services": {"enabled_removed": ["nginx.service"]}}
|
||||
assert has_enforceable_drift(report) is True
|
||||
|
||||
|
||||
def test_has_enforceable_drift_service_changed():
|
||||
report = {
|
||||
"services": {
|
||||
"changed": [
|
||||
{
|
||||
"unit": "nginx.service",
|
||||
"changes": {"active_state": {"old": "active", "new": "inactive"}},
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
assert has_enforceable_drift(report) is True
|
||||
|
||||
|
||||
def test_has_enforceable_drift_service_package_only_changed():
|
||||
# Service changed only in packages - should NOT be enforceable
|
||||
report = {
|
||||
"services": {
|
||||
"changed": [
|
||||
{
|
||||
"unit": "nginx.service",
|
||||
"changes": {"packages": {"added": ["nginx-extra"]}},
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
assert has_enforceable_drift(report) is False
|
||||
|
||||
|
||||
def test_has_enforceable_drift_users_removed():
|
||||
report = {"users": {"removed": ["alice"]}}
|
||||
assert has_enforceable_drift(report) is True
|
||||
|
||||
|
||||
def test_has_enforceable_drift_users_changed():
|
||||
report = {
|
||||
"users": {
|
||||
"changed": [
|
||||
{"name": "alice", "changes": {"uid": {"old": 1000, "new": 1001}}}
|
||||
]
|
||||
}
|
||||
}
|
||||
assert has_enforceable_drift(report) is True
|
||||
|
||||
|
||||
def test_has_enforceable_drift_files_removed():
|
||||
report = {
|
||||
"files": {
|
||||
"removed": [{"path": "/etc/passwd", "role": "users", "reason": "conffile"}]
|
||||
}
|
||||
}
|
||||
assert has_enforceable_drift(report) is True
|
||||
|
||||
|
||||
def test_has_enforceable_drift_files_changed():
|
||||
report = {
|
||||
"files": {
|
||||
"changed": [
|
||||
{
|
||||
"path": "/etc/passwd",
|
||||
"changes": {"content": {"old": "sha1", "new": "sha2"}},
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
assert has_enforceable_drift(report) is True
|
||||
|
||||
|
||||
def test_has_enforceable_drift_no_drift():
|
||||
report = {
|
||||
"packages": {"added": ["newpkg"]},
|
||||
"services": {"enabled_added": ["new.service"]},
|
||||
"users": {"added": ["bob"]},
|
||||
"files": {"added": ["/opt/newfile"]},
|
||||
}
|
||||
assert has_enforceable_drift(report) is False
|
||||
|
||||
|
||||
def test_enforcement_plan_packages_removed(monkeypatch, tmp_path: Path):
|
||||
old_state = {
|
||||
"roles": {
|
||||
"services": [{"role_name": "nginx", "packages": ["nginx"]}],
|
||||
"packages": [{"role_name": "vim", "package": "vim"}],
|
||||
}
|
||||
}
|
||||
report = {"packages": {"removed": ["nginx", "vim"]}}
|
||||
|
||||
result = _enforcement_plan(report, old_state, tmp_path)
|
||||
assert "nginx" in result.get("roles", [])
|
||||
assert "vim" in result.get("roles", [])
|
||||
assert "role_nginx" in result.get("tags", [])
|
||||
|
||||
|
||||
def test_enforcement_plan_users_changed():
|
||||
old_state = {
|
||||
"roles": {"users": {"role_name": "users", "users": [{"name": "alice"}]}}
|
||||
}
|
||||
report = {"users": {"changed": [{"name": "alice", "changes": {"uid": {}}}]}}
|
||||
|
||||
result = _enforcement_plan(report, old_state, Path("/tmp"))
|
||||
assert "users" in result.get("roles", [])
|
||||
|
||||
|
||||
def test_enforcement_plan_files_removed(tmp_path: Path):
|
||||
# Create the artifacts directory structure that _file_index expects
|
||||
artifacts_dir = tmp_path / "artifacts" / "etc_custom"
|
||||
artifacts_dir.mkdir(parents=True)
|
||||
|
||||
old_state = {
|
||||
"roles": {
|
||||
"etc_custom": {
|
||||
"role_name": "etc_custom",
|
||||
"managed_files": [
|
||||
{"path": "/etc/custom.conf", "src_rel": "custom.conf"}
|
||||
],
|
||||
}
|
||||
}
|
||||
}
|
||||
report = {
|
||||
"files": {"removed": [{"path": "/etc/custom.conf", "role": "etc_custom"}]}
|
||||
}
|
||||
|
||||
result = _enforcement_plan(report, old_state, tmp_path)
|
||||
assert "etc_custom" in result.get("roles", [])
|
||||
|
||||
|
||||
def test_enforcement_plan_no_drift():
|
||||
old_state = {"roles": {}}
|
||||
report = {"packages": {"added": ["newpkg"]}}
|
||||
|
||||
result = _enforcement_plan(report, old_state, Path("/tmp"))
|
||||
assert result.get("roles", []) == []
|
||||
|
||||
|
||||
def test_bundle_from_input_tgz(monkeypatch, tmp_path: Path):
|
||||
bundle_dir = tmp_path / "bundle"
|
||||
bundle_dir.mkdir()
|
||||
|
|
@ -989,64 +843,6 @@ def test_report_markdown_basic():
|
|||
assert "+ vim" in result
|
||||
|
||||
|
||||
def test_report_markdown_with_enforcement_applied():
|
||||
report = {
|
||||
"generated_at": "2024-01-01T00:00:00Z",
|
||||
"old": {"input": "old.tar.gz"},
|
||||
"new": {"input": "new.tar.gz"},
|
||||
"packages": {"added": [], "removed": [], "version_changed": []},
|
||||
"services": {"enabled_added": [], "enabled_removed": [], "changed": []},
|
||||
"users": {"added": [], "removed": [], "changed": []},
|
||||
"files": {"added": [], "removed": [], "changed": []},
|
||||
"enforcement": {
|
||||
"status": "applied",
|
||||
"tags": ["role_users"],
|
||||
"returncode": 0,
|
||||
"finished_at": "2024-01-01T00:01:00Z",
|
||||
},
|
||||
}
|
||||
result = _report_markdown(report)
|
||||
assert "Applied old harvest" in result
|
||||
assert "role_users" in result
|
||||
|
||||
|
||||
def test_report_markdown_with_enforcement_failed():
|
||||
report = {
|
||||
"generated_at": "2024-01-01T00:00:00Z",
|
||||
"old": {"input": "old.tar.gz"},
|
||||
"new": {"input": "new.tar.gz"},
|
||||
"packages": {"added": [], "removed": [], "version_changed": []},
|
||||
"services": {"enabled_added": [], "enabled_removed": [], "changed": []},
|
||||
"users": {"added": [], "removed": [], "changed": []},
|
||||
"files": {"added": [], "removed": [], "changed": []},
|
||||
"enforcement": {
|
||||
"status": "failed",
|
||||
"returncode": 1,
|
||||
},
|
||||
}
|
||||
result = _report_markdown(report)
|
||||
assert "but failed" in result
|
||||
|
||||
|
||||
def test_report_markdown_with_enforcement_skipped():
|
||||
report = {
|
||||
"generated_at": "2024-01-01T00:00:00Z",
|
||||
"old": {"input": "old.tar.gz"},
|
||||
"new": {"input": "new.tar.gz"},
|
||||
"packages": {"added": [], "removed": [], "version_changed": []},
|
||||
"services": {"enabled_added": [], "enabled_removed": [], "changed": []},
|
||||
"users": {"added": [], "removed": [], "changed": []},
|
||||
"files": {"added": [], "removed": [], "changed": []},
|
||||
"enforcement": {
|
||||
"status": "skipped",
|
||||
"reason": "no drift",
|
||||
},
|
||||
}
|
||||
result = _report_markdown(report)
|
||||
assert "Skipped" in result
|
||||
assert "no drift" in result
|
||||
|
||||
|
||||
def test_report_markdown_with_version_ignored():
|
||||
report = {
|
||||
"generated_at": "2024-01-01T00:00:00Z",
|
||||
|
|
@ -1355,67 +1151,3 @@ def test_report_text_with_ignore_package_versions():
|
|||
result = d._report_text(report)
|
||||
assert "package version drift: ignored" in result
|
||||
assert "ignored 5 changes" in result
|
||||
|
||||
|
||||
def test_report_text_with_enforcement_applied():
|
||||
"""Test _report_text includes enforcement applied status."""
|
||||
import enroll.diff as d
|
||||
|
||||
report = {
|
||||
"generated_at": "2024-01-01T00:00:00Z",
|
||||
"old": {"input": "old.tar.gz", "host": "host1", "state_mtime": "mtime1"},
|
||||
"new": {"input": "new.tar.gz", "host": "host2", "state_mtime": "mtime2"},
|
||||
"packages": {"added": [], "removed": [], "version_changed": []},
|
||||
"services": {"enabled_added": [], "enabled_removed": [], "changed": []},
|
||||
"users": {"added": [], "removed": [], "changed": []},
|
||||
"files": {"added": [], "removed": [], "changed": []},
|
||||
"enforcement": {
|
||||
"status": "applied",
|
||||
"returncode": 0,
|
||||
"tags": ["test"],
|
||||
"finished_at": "2024-01-01T01:00:00Z",
|
||||
},
|
||||
}
|
||||
result = d._report_text(report)
|
||||
assert "Enforcement" in result
|
||||
assert "applied old harvest" in result
|
||||
assert "tags=test" in result
|
||||
|
||||
|
||||
def test_report_text_with_enforcement_failed():
|
||||
"""Test _report_text includes enforcement failed status."""
|
||||
import enroll.diff as d
|
||||
|
||||
report = {
|
||||
"generated_at": "2024-01-01T00:00:00Z",
|
||||
"old": {"input": "old.tar.gz", "host": "host1", "state_mtime": "mtime1"},
|
||||
"new": {"input": "new.tar.gz", "host": "host2", "state_mtime": "mtime2"},
|
||||
"packages": {"added": [], "removed": [], "version_changed": []},
|
||||
"services": {"enabled_added": [], "enabled_removed": [], "changed": []},
|
||||
"users": {"added": [], "removed": [], "changed": []},
|
||||
"files": {"added": [], "removed": [], "changed": []},
|
||||
"enforcement": {"status": "failed", "returncode": 1},
|
||||
}
|
||||
result = d._report_text(report)
|
||||
assert "Enforcement" in result
|
||||
assert "failed" in result
|
||||
|
||||
|
||||
def test_report_text_with_enforcement_skipped():
|
||||
"""Test _report_text includes enforcement skipped status."""
|
||||
import enroll.diff as d
|
||||
|
||||
report = {
|
||||
"generated_at": "2024-01-01T00:00:00Z",
|
||||
"old": {"input": "old.tar.gz", "host": "host1", "state_mtime": "mtime1"},
|
||||
"new": {"input": "new.tar.gz", "host": "host2", "state_mtime": "mtime2"},
|
||||
"packages": {"added": [], "removed": [], "version_changed": []},
|
||||
"services": {"enabled_added": [], "enabled_removed": [], "changed": []},
|
||||
"users": {"added": [], "removed": [], "changed": []},
|
||||
"files": {"added": [], "removed": [], "changed": []},
|
||||
"enforcement": {"status": "skipped", "reason": "no changes"},
|
||||
}
|
||||
result = d._report_text(report)
|
||||
assert "Enforcement" in result
|
||||
assert "skipped" in result
|
||||
assert "no changes" in result
|
||||
|
|
|
|||
214
tests/test_diff_ignore_versions_exclude.py
Normal file
214
tests/test_diff_ignore_versions_exclude.py
Normal file
|
|
@ -0,0 +1,214 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _write_bundle(
|
||||
root: Path, state: dict, artifacts: dict[str, bytes] | None = None
|
||||
) -> None:
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
(root / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
|
||||
artifacts = artifacts or {}
|
||||
for rel, data in artifacts.items():
|
||||
p = root / rel
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_bytes(data)
|
||||
|
||||
|
||||
def _minimal_roles() -> dict:
|
||||
"""A small roles structure that's sufficient for enroll.diff file indexing."""
|
||||
return {
|
||||
"users": {
|
||||
"role_name": "users",
|
||||
"users": [],
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
"services": [],
|
||||
"packages": [],
|
||||
"apt_config": {
|
||||
"role_name": "apt_config",
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
"etc_custom": {
|
||||
"role_name": "etc_custom",
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
"usr_local_custom": {
|
||||
"role_name": "usr_local_custom",
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
"extra_paths": {
|
||||
"role_name": "extra_paths",
|
||||
"include_patterns": [],
|
||||
"exclude_patterns": [],
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_diff_ignore_package_versions_suppresses_version_drift(tmp_path: Path):
|
||||
from enroll.diff import compare_harvests
|
||||
|
||||
old = tmp_path / "old"
|
||||
new = tmp_path / "new"
|
||||
|
||||
old_state = {
|
||||
"schema_version": 3,
|
||||
"host": {"hostname": "h1"},
|
||||
"inventory": {
|
||||
"packages": {
|
||||
"curl": {
|
||||
"version": "1.0",
|
||||
"installations": [{"version": "1.0", "arch": "amd64"}],
|
||||
}
|
||||
}
|
||||
},
|
||||
"roles": _minimal_roles(),
|
||||
}
|
||||
new_state = {
|
||||
**old_state,
|
||||
"inventory": {
|
||||
"packages": {
|
||||
"curl": {
|
||||
"version": "1.1",
|
||||
"installations": [{"version": "1.1", "arch": "amd64"}],
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
_write_bundle(old, old_state)
|
||||
_write_bundle(new, new_state)
|
||||
|
||||
# Without ignore flag, version drift is reported and counts as changes.
|
||||
report, has_changes = compare_harvests(str(old), str(new))
|
||||
assert has_changes is True
|
||||
assert report["packages"]["version_changed"]
|
||||
|
||||
# With ignore flag, version drift is suppressed and does not count as changes.
|
||||
report2, has_changes2 = compare_harvests(
|
||||
str(old), str(new), ignore_package_versions=True
|
||||
)
|
||||
assert has_changes2 is False
|
||||
assert report2["packages"]["version_changed"] == []
|
||||
assert report2["packages"]["version_changed_ignored_count"] == 1
|
||||
assert report2["filters"]["ignore_package_versions"] is True
|
||||
|
||||
|
||||
def test_diff_exclude_path_filters_file_drift_and_affects_has_changes(tmp_path: Path):
|
||||
from enroll.diff import compare_harvests
|
||||
|
||||
old = tmp_path / "old"
|
||||
new = tmp_path / "new"
|
||||
|
||||
# Only file drift is under /var/anacron, which is excluded.
|
||||
old_state = {
|
||||
"schema_version": 3,
|
||||
"host": {"hostname": "h1"},
|
||||
"inventory": {"packages": {}},
|
||||
"roles": {
|
||||
**_minimal_roles(),
|
||||
"extra_paths": {
|
||||
**_minimal_roles()["extra_paths"],
|
||||
"managed_files": [
|
||||
{
|
||||
"path": "/var/anacron/daily.stamp",
|
||||
"src_rel": "var/anacron/daily.stamp",
|
||||
"owner": "root",
|
||||
"group": "root",
|
||||
"mode": "0644",
|
||||
"reason": "extra_path",
|
||||
}
|
||||
],
|
||||
},
|
||||
},
|
||||
}
|
||||
new_state = json.loads(json.dumps(old_state))
|
||||
|
||||
_write_bundle(
|
||||
old,
|
||||
old_state,
|
||||
{"artifacts/extra_paths/var/anacron/daily.stamp": b"yesterday\n"},
|
||||
)
|
||||
_write_bundle(
|
||||
new,
|
||||
new_state,
|
||||
{"artifacts/extra_paths/var/anacron/daily.stamp": b"today\n"},
|
||||
)
|
||||
|
||||
report, has_changes = compare_harvests(
|
||||
str(old), str(new), exclude_paths=["/var/anacron"]
|
||||
)
|
||||
assert has_changes is False
|
||||
assert report["files"]["changed"] == []
|
||||
assert report["filters"]["exclude_paths"] == ["/var/anacron"]
|
||||
|
||||
|
||||
def test_diff_exclude_path_only_filters_files_not_packages(tmp_path: Path):
|
||||
from enroll.diff import compare_harvests
|
||||
|
||||
old = tmp_path / "old"
|
||||
new = tmp_path / "new"
|
||||
|
||||
old_state = {
|
||||
"schema_version": 3,
|
||||
"host": {"hostname": "h1"},
|
||||
"inventory": {"packages": {"curl": {"version": "1.0"}}},
|
||||
"roles": {
|
||||
**_minimal_roles(),
|
||||
"extra_paths": {
|
||||
**_minimal_roles()["extra_paths"],
|
||||
"managed_files": [
|
||||
{
|
||||
"path": "/var/anacron/daily.stamp",
|
||||
"src_rel": "var/anacron/daily.stamp",
|
||||
"owner": "root",
|
||||
"group": "root",
|
||||
"mode": "0644",
|
||||
"reason": "extra_path",
|
||||
}
|
||||
],
|
||||
},
|
||||
},
|
||||
}
|
||||
new_state = {
|
||||
**old_state,
|
||||
"inventory": {
|
||||
"packages": {
|
||||
"curl": {"version": "1.0"},
|
||||
"htop": {"version": "3.0"},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
_write_bundle(
|
||||
old,
|
||||
old_state,
|
||||
{"artifacts/extra_paths/var/anacron/daily.stamp": b"yesterday\n"},
|
||||
)
|
||||
_write_bundle(
|
||||
new,
|
||||
new_state,
|
||||
{
|
||||
"artifacts/extra_paths/var/anacron/daily.stamp": b"today\n",
|
||||
},
|
||||
)
|
||||
|
||||
report, has_changes = compare_harvests(
|
||||
str(old), str(new), exclude_paths=["/var/anacron"]
|
||||
)
|
||||
assert has_changes is True
|
||||
# File drift is filtered, but package drift remains.
|
||||
assert report["files"]["changed"] == []
|
||||
assert report["packages"]["added"] == ["htop"]
|
||||
|
|
@ -1,454 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import types
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _write_bundle(
|
||||
root: Path, state: dict, artifacts: dict[str, bytes] | None = None
|
||||
) -> None:
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
(root / "state.json").write_text(json.dumps(state, indent=2), encoding="utf-8")
|
||||
artifacts = artifacts or {}
|
||||
for rel, data in artifacts.items():
|
||||
p = root / rel
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_bytes(data)
|
||||
|
||||
|
||||
def _minimal_roles() -> dict:
|
||||
"""A small roles structure that's sufficient for enroll.diff file indexing."""
|
||||
return {
|
||||
"users": {
|
||||
"role_name": "users",
|
||||
"users": [],
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
"services": [],
|
||||
"packages": [],
|
||||
"apt_config": {
|
||||
"role_name": "apt_config",
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
"etc_custom": {
|
||||
"role_name": "etc_custom",
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
"usr_local_custom": {
|
||||
"role_name": "usr_local_custom",
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
"extra_paths": {
|
||||
"role_name": "extra_paths",
|
||||
"include_patterns": [],
|
||||
"exclude_patterns": [],
|
||||
"managed_files": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_diff_ignore_package_versions_suppresses_version_drift(tmp_path: Path):
|
||||
from enroll.diff import compare_harvests
|
||||
|
||||
old = tmp_path / "old"
|
||||
new = tmp_path / "new"
|
||||
|
||||
old_state = {
|
||||
"schema_version": 3,
|
||||
"host": {"hostname": "h1"},
|
||||
"inventory": {
|
||||
"packages": {
|
||||
"curl": {
|
||||
"version": "1.0",
|
||||
"installations": [{"version": "1.0", "arch": "amd64"}],
|
||||
}
|
||||
}
|
||||
},
|
||||
"roles": _minimal_roles(),
|
||||
}
|
||||
new_state = {
|
||||
**old_state,
|
||||
"inventory": {
|
||||
"packages": {
|
||||
"curl": {
|
||||
"version": "1.1",
|
||||
"installations": [{"version": "1.1", "arch": "amd64"}],
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
_write_bundle(old, old_state)
|
||||
_write_bundle(new, new_state)
|
||||
|
||||
# Without ignore flag, version drift is reported and counts as changes.
|
||||
report, has_changes = compare_harvests(str(old), str(new))
|
||||
assert has_changes is True
|
||||
assert report["packages"]["version_changed"]
|
||||
|
||||
# With ignore flag, version drift is suppressed and does not count as changes.
|
||||
report2, has_changes2 = compare_harvests(
|
||||
str(old), str(new), ignore_package_versions=True
|
||||
)
|
||||
assert has_changes2 is False
|
||||
assert report2["packages"]["version_changed"] == []
|
||||
assert report2["packages"]["version_changed_ignored_count"] == 1
|
||||
assert report2["filters"]["ignore_package_versions"] is True
|
||||
|
||||
|
||||
def test_diff_exclude_path_filters_file_drift_and_affects_has_changes(tmp_path: Path):
|
||||
from enroll.diff import compare_harvests
|
||||
|
||||
old = tmp_path / "old"
|
||||
new = tmp_path / "new"
|
||||
|
||||
# Only file drift is under /var/anacron, which is excluded.
|
||||
old_state = {
|
||||
"schema_version": 3,
|
||||
"host": {"hostname": "h1"},
|
||||
"inventory": {"packages": {}},
|
||||
"roles": {
|
||||
**_minimal_roles(),
|
||||
"extra_paths": {
|
||||
**_minimal_roles()["extra_paths"],
|
||||
"managed_files": [
|
||||
{
|
||||
"path": "/var/anacron/daily.stamp",
|
||||
"src_rel": "var/anacron/daily.stamp",
|
||||
"owner": "root",
|
||||
"group": "root",
|
||||
"mode": "0644",
|
||||
"reason": "extra_path",
|
||||
}
|
||||
],
|
||||
},
|
||||
},
|
||||
}
|
||||
new_state = json.loads(json.dumps(old_state))
|
||||
|
||||
_write_bundle(
|
||||
old,
|
||||
old_state,
|
||||
{"artifacts/extra_paths/var/anacron/daily.stamp": b"yesterday\n"},
|
||||
)
|
||||
_write_bundle(
|
||||
new,
|
||||
new_state,
|
||||
{"artifacts/extra_paths/var/anacron/daily.stamp": b"today\n"},
|
||||
)
|
||||
|
||||
report, has_changes = compare_harvests(
|
||||
str(old), str(new), exclude_paths=["/var/anacron"]
|
||||
)
|
||||
assert has_changes is False
|
||||
assert report["files"]["changed"] == []
|
||||
assert report["filters"]["exclude_paths"] == ["/var/anacron"]
|
||||
|
||||
|
||||
def test_diff_exclude_path_only_filters_files_not_packages(tmp_path: Path):
|
||||
from enroll.diff import compare_harvests
|
||||
|
||||
old = tmp_path / "old"
|
||||
new = tmp_path / "new"
|
||||
|
||||
old_state = {
|
||||
"schema_version": 3,
|
||||
"host": {"hostname": "h1"},
|
||||
"inventory": {"packages": {"curl": {"version": "1.0"}}},
|
||||
"roles": {
|
||||
**_minimal_roles(),
|
||||
"extra_paths": {
|
||||
**_minimal_roles()["extra_paths"],
|
||||
"managed_files": [
|
||||
{
|
||||
"path": "/var/anacron/daily.stamp",
|
||||
"src_rel": "var/anacron/daily.stamp",
|
||||
"owner": "root",
|
||||
"group": "root",
|
||||
"mode": "0644",
|
||||
"reason": "extra_path",
|
||||
}
|
||||
],
|
||||
},
|
||||
},
|
||||
}
|
||||
new_state = {
|
||||
**old_state,
|
||||
"inventory": {
|
||||
"packages": {
|
||||
"curl": {"version": "1.0"},
|
||||
"htop": {"version": "3.0"},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
_write_bundle(
|
||||
old,
|
||||
old_state,
|
||||
{"artifacts/extra_paths/var/anacron/daily.stamp": b"yesterday\n"},
|
||||
)
|
||||
_write_bundle(
|
||||
new,
|
||||
new_state,
|
||||
{
|
||||
"artifacts/extra_paths/var/anacron/daily.stamp": b"today\n",
|
||||
},
|
||||
)
|
||||
|
||||
report, has_changes = compare_harvests(
|
||||
str(old), str(new), exclude_paths=["/var/anacron"]
|
||||
)
|
||||
assert has_changes is True
|
||||
# File drift is filtered, but package drift remains.
|
||||
assert report["files"]["changed"] == []
|
||||
assert report["packages"]["added"] == ["htop"]
|
||||
|
||||
|
||||
def test_enforce_old_harvest_runs_ansible_with_tags_from_file_drift(
|
||||
monkeypatch, tmp_path: Path
|
||||
):
|
||||
import enroll.diff as d
|
||||
import enroll.manifest as mf
|
||||
|
||||
# Pretend ansible-playbook is installed.
|
||||
monkeypatch.setattr(d.shutil, "which", lambda name: "/usr/bin/ansible-playbook")
|
||||
|
||||
calls: dict[str, object] = {}
|
||||
|
||||
# Stub manifest generation to only create playbook.yml (fast, no real roles needed).
|
||||
def fake_manifest(_harvest_dir: str, out_dir: str, **_kwargs):
|
||||
out = Path(out_dir)
|
||||
out.mkdir(parents=True, exist_ok=False)
|
||||
(out / "playbook.yml").write_text(
|
||||
"---\n- hosts: all\n gather_facts: false\n roles: []\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
monkeypatch.setattr(mf, "manifest", fake_manifest)
|
||||
|
||||
def fake_run(
|
||||
argv, cwd=None, env=None, capture_output=False, text=False, check=False
|
||||
):
|
||||
calls["argv"] = list(argv)
|
||||
calls["cwd"] = cwd
|
||||
return types.SimpleNamespace(returncode=0, stdout="ok", stderr="")
|
||||
|
||||
monkeypatch.setattr(d.subprocess, "run", fake_run)
|
||||
|
||||
old = tmp_path / "old"
|
||||
old_state = {
|
||||
"schema_version": 3,
|
||||
"host": {"hostname": "h1"},
|
||||
"inventory": {"packages": {}},
|
||||
"roles": {
|
||||
**_minimal_roles(),
|
||||
"usr_local_custom": {
|
||||
**_minimal_roles()["usr_local_custom"],
|
||||
"managed_files": [
|
||||
{
|
||||
"path": "/etc/myapp.conf",
|
||||
"src_rel": "etc/myapp.conf",
|
||||
"owner": "root",
|
||||
"group": "root",
|
||||
"mode": "0644",
|
||||
"reason": "custom",
|
||||
}
|
||||
],
|
||||
},
|
||||
},
|
||||
}
|
||||
_write_bundle(old, old_state)
|
||||
|
||||
# Minimal report containing enforceable drift: a baseline file is "removed".
|
||||
report = {
|
||||
"packages": {"added": [], "removed": [], "version_changed": []},
|
||||
"services": {"enabled_added": [], "enabled_removed": [], "changed": []},
|
||||
"users": {"added": [], "removed": [], "changed": []},
|
||||
"files": {
|
||||
"added": [],
|
||||
"removed": [{"path": "/etc/myapp.conf", "role": "usr_local_custom"}],
|
||||
"changed": [],
|
||||
},
|
||||
}
|
||||
|
||||
info = d.enforce_old_harvest(str(old), report=report)
|
||||
assert info["status"] == "applied"
|
||||
assert "--tags" in info["command"]
|
||||
assert "role_usr_local_custom" in ",".join(info.get("tags") or [])
|
||||
|
||||
argv = calls.get("argv")
|
||||
assert argv and argv[0].endswith("ansible-playbook")
|
||||
assert "--tags" in argv
|
||||
# Ensure we pass the computed tag.
|
||||
i = argv.index("--tags")
|
||||
assert "role_usr_local_custom" in str(argv[i + 1])
|
||||
|
||||
|
||||
def test_cli_diff_enforce_forwards_target(monkeypatch):
|
||||
import enroll.cli as cli
|
||||
|
||||
report = {
|
||||
"packages": {"added": [], "removed": ["curl"], "version_changed": []},
|
||||
"services": {"enabled_added": [], "enabled_removed": [], "changed": []},
|
||||
"users": {"added": [], "removed": [], "changed": []},
|
||||
"files": {"added": [], "removed": [], "changed": []},
|
||||
}
|
||||
|
||||
monkeypatch.setattr(cli, "compare_harvests", lambda *a, **k: (report, True))
|
||||
monkeypatch.setattr(cli, "has_enforceable_drift", lambda r: True)
|
||||
|
||||
calls: dict[str, object] = {}
|
||||
|
||||
def fake_enforce(old, **kwargs):
|
||||
calls["old"] = old
|
||||
calls.update(kwargs)
|
||||
return {"status": "applied", "target": kwargs.get("target"), "returncode": 0}
|
||||
|
||||
monkeypatch.setattr(cli, "enforce_old_harvest", fake_enforce)
|
||||
monkeypatch.setattr(cli, "format_report", lambda report, fmt="text": "R\n")
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
[
|
||||
"enroll",
|
||||
"diff",
|
||||
"--old",
|
||||
"/tmp/old",
|
||||
"--new",
|
||||
"/tmp/new",
|
||||
"--enforce",
|
||||
],
|
||||
)
|
||||
|
||||
cli.main()
|
||||
assert calls["old"] == "/tmp/old"
|
||||
assert calls["report"] is report
|
||||
|
||||
|
||||
def test_cli_diff_enforce_rejects_non_ansible_target(monkeypatch):
|
||||
"""Salt and Puppet targets have been removed from --enforce."""
|
||||
import enroll.cli as cli
|
||||
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
[
|
||||
"enroll",
|
||||
"diff",
|
||||
"--old",
|
||||
"/tmp/old",
|
||||
"--new",
|
||||
"/tmp/new",
|
||||
"--enforce",
|
||||
"--target",
|
||||
"puppet",
|
||||
],
|
||||
)
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli.main()
|
||||
|
||||
|
||||
def test_cli_diff_forwards_exclude_and_ignore_flags(monkeypatch, capsys):
|
||||
import enroll.cli as cli
|
||||
|
||||
calls: dict[str, object] = {}
|
||||
|
||||
def fake_compare(
|
||||
old, new, *, sops_mode=False, exclude_paths=None, ignore_package_versions=False
|
||||
):
|
||||
calls["compare"] = {
|
||||
"old": old,
|
||||
"new": new,
|
||||
"sops_mode": sops_mode,
|
||||
"exclude_paths": exclude_paths,
|
||||
"ignore_package_versions": ignore_package_versions,
|
||||
}
|
||||
# No changes -> should not try to enforce.
|
||||
return {"packages": {}, "services": {}, "users": {}, "files": {}}, False
|
||||
|
||||
monkeypatch.setattr(cli, "compare_harvests", fake_compare)
|
||||
monkeypatch.setattr(cli, "format_report", lambda report, fmt="text": "R\n")
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
[
|
||||
"enroll",
|
||||
"diff",
|
||||
"--old",
|
||||
"/tmp/old",
|
||||
"--new",
|
||||
"/tmp/new",
|
||||
"--exclude-path",
|
||||
"/var/anacron",
|
||||
"--ignore-package-versions",
|
||||
],
|
||||
)
|
||||
|
||||
cli.main()
|
||||
_ = capsys.readouterr()
|
||||
assert calls["compare"]["exclude_paths"] == ["/var/anacron"]
|
||||
assert calls["compare"]["ignore_package_versions"] is True
|
||||
|
||||
|
||||
def test_cli_diff_enforce_skips_when_no_enforceable_drift(monkeypatch):
|
||||
import enroll.cli as cli
|
||||
|
||||
# Drift exists, but is not enforceable (only additions / version changes).
|
||||
report = {
|
||||
"packages": {"added": ["htop"], "removed": [], "version_changed": []},
|
||||
"services": {
|
||||
"enabled_added": ["x.service"],
|
||||
"enabled_removed": [],
|
||||
"changed": [],
|
||||
},
|
||||
"users": {"added": ["bob"], "removed": [], "changed": []},
|
||||
"files": {"added": [{"path": "/tmp/new"}], "removed": [], "changed": []},
|
||||
}
|
||||
|
||||
monkeypatch.setattr(cli, "compare_harvests", lambda *a, **k: (report, True))
|
||||
monkeypatch.setattr(cli, "has_enforceable_drift", lambda r: False)
|
||||
called = {"enforce": False}
|
||||
monkeypatch.setattr(
|
||||
cli, "enforce_old_harvest", lambda *a, **k: called.update({"enforce": True})
|
||||
)
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_format(rep, fmt="text"):
|
||||
captured["report"] = rep
|
||||
return "R\n"
|
||||
|
||||
monkeypatch.setattr(cli, "format_report", fake_format)
|
||||
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
[
|
||||
"enroll",
|
||||
"diff",
|
||||
"--old",
|
||||
"/tmp/old",
|
||||
"--new",
|
||||
"/tmp/new",
|
||||
"--enforce",
|
||||
],
|
||||
)
|
||||
|
||||
cli.main()
|
||||
assert called["enforce"] is False
|
||||
assert captured["report"]["enforcement"]["status"] == "skipped"
|
||||
|
|
@ -637,15 +637,23 @@ def test_manifest_groups_systemd_units_into_common_role(tmp_path: Path):
|
|||
assert "Ensure grouped unit enablement matches harvest" in tasks
|
||||
assert 'no_log: "{{ enroll_hide_systemd_status | default(true) | bool }}"' in tasks
|
||||
assert "enroll_manage_systemd_runtime | default(true) | bool" in tasks
|
||||
assert "Restart managed services" not in tasks
|
||||
|
||||
defaults_text = (out / "roles" / "net" / "defaults" / "main.yml").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
# Notify points at the role's fixed restart topic (scaffold-safe), never a
|
||||
# per-unit handler name built from harvested data.
|
||||
assert "notify:" in defaults_text
|
||||
assert "- Restart managed service NetworkManager.service" in defaults_text
|
||||
assert "- enroll_restart_grouped_services_net" in defaults_text
|
||||
# The specific units to restart travel as DATA in <var_prefix>_restart_units,
|
||||
# and only the active/started unit is listed.
|
||||
assert "net_restart_units:" in defaults_text
|
||||
assert "- NetworkManager.service" in defaults_text
|
||||
assert (
|
||||
"Restart managed service NetworkManager-dispatcher.service" not in defaults_text
|
||||
"NetworkManager-dispatcher.service"
|
||||
not in defaults_text.split("net_restart_units:")[1].split("net_systemd_units:")[
|
||||
0
|
||||
]
|
||||
)
|
||||
|
||||
handlers = (out / "roles" / "net" / "handlers" / "main.yml").read_text(
|
||||
|
|
@ -653,11 +661,15 @@ def test_manifest_groups_systemd_units_into_common_role(tmp_path: Path):
|
|||
)
|
||||
assert "Run systemd daemon-reload" in handlers
|
||||
assert "when: enroll_manage_systemd_runtime | default(true) | bool" in handlers
|
||||
assert "- name: Restart managed service NetworkManager.service" in handlers
|
||||
assert "name: NetworkManager.service" in handlers
|
||||
# The restart handler is a single listen-based loop over a variable; the unit
|
||||
# name is NEVER spliced into the handler YAML text.
|
||||
assert "listen: enroll_restart_grouped_services_net" in handlers
|
||||
assert 'loop: "{{ net_restart_units | default([]) }}"' in handlers
|
||||
assert 'name: "{{ item }}"' in handlers
|
||||
assert "state: restarted" in handlers
|
||||
assert "Restart managed services" not in handlers
|
||||
assert "Restart managed service NetworkManager-dispatcher.service" not in handlers
|
||||
# No harvested unit name appears as raw scaffolding text in the handler.
|
||||
assert "NetworkManager.service" not in handlers
|
||||
assert "NetworkManager-dispatcher.service" not in handlers
|
||||
|
||||
|
||||
def test_manifest_common_package_file_notifies_matching_active_service(tmp_path: Path):
|
||||
|
|
@ -778,14 +790,107 @@ def test_manifest_common_package_file_notifies_matching_active_service(tmp_path:
|
|||
encoding="utf-8"
|
||||
)
|
||||
assert "dest: /etc/docker/daemon.json" in defaults
|
||||
assert "- Restart managed service docker.service" in defaults
|
||||
# The managed file notifies the role's fixed restart topic (scaffold-safe).
|
||||
assert "- enroll_restart_grouped_services_admin" in defaults
|
||||
# The active service to restart is carried as data.
|
||||
assert "admin_restart_units:" in defaults
|
||||
assert "- docker.service" in defaults
|
||||
|
||||
handlers = (out / "roles" / "admin" / "handlers" / "main.yml").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
assert "- name: Restart managed service docker.service" in handlers
|
||||
assert "name: docker.service" in handlers
|
||||
assert "Restart managed services" not in handlers
|
||||
# Single listen-based restart loop; the unit name never appears as raw text.
|
||||
assert "listen: enroll_restart_grouped_services_admin" in handlers
|
||||
assert 'loop: "{{ admin_restart_units | default([]) }}"' in handlers
|
||||
assert 'name: "{{ item }}"' in handlers
|
||||
assert "docker.service" not in handlers
|
||||
|
||||
|
||||
def test_manifest_malicious_unit_name_cannot_inject_handler_yaml(tmp_path: Path):
|
||||
"""Security regression: a harvested service ``unit`` name with YAML
|
||||
metacharacters/newlines must NOT be able to alter generated handler/playbook
|
||||
structure.
|
||||
|
||||
Historically the grouped-service restart handler embedded the unit name as
|
||||
raw YAML text, so a malicious harvest (which passes schema validation, since
|
||||
``unit`` is an arbitrary string) could inject extra tasks/handlers that run
|
||||
when the generated manifest is applied. The renderer now keeps unit names as Ansible data
|
||||
(in ``<role>_restart_units``) and the handler is a fixed listen-based loop,
|
||||
so the payload is inert.
|
||||
"""
|
||||
|
||||
import yaml
|
||||
|
||||
bundle = tmp_path / "bundle"
|
||||
out = tmp_path / "ansible"
|
||||
|
||||
payload_unit = (
|
||||
"evil.service\n"
|
||||
" ansible.builtin.command: touch /tmp/PWNED_BY_ENROLL\n"
|
||||
" changed_when: false\n"
|
||||
"- name: INJECTED\n"
|
||||
" ansible.builtin.command: id\n"
|
||||
)
|
||||
|
||||
state = {
|
||||
"host": {"hostname": "test", "os": "debian", "pkg_backend": "dpkg"},
|
||||
"roles": {
|
||||
"services": [
|
||||
{
|
||||
"unit": payload_unit,
|
||||
"role_name": "evilrole",
|
||||
"packages": [],
|
||||
"active_state": "active",
|
||||
"sub_state": "running",
|
||||
"unit_file_state": "enabled",
|
||||
"condition_result": "yes",
|
||||
"managed_files": [],
|
||||
"managed_dirs": [],
|
||||
"managed_links": [],
|
||||
"excluded": [],
|
||||
"notes": [],
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
_write_state(bundle, state)
|
||||
|
||||
# Must render without raising and without producing structurally-injected YAML.
|
||||
manifest.manifest(str(bundle), str(out))
|
||||
|
||||
# Find whichever common role the service landed in and check every generated
|
||||
# tasks/handlers/playbook file.
|
||||
yml_files = list((out).rglob("handlers/main.yml"))
|
||||
yml_files += list((out).rglob("tasks/main.yml"))
|
||||
yml_files += [p for p in (out).rglob("*.yml") if p.name == "playbook.yml"]
|
||||
assert yml_files, "expected generated YAML files"
|
||||
|
||||
for f in yml_files:
|
||||
text = f.read_text(encoding="utf-8")
|
||||
# The injected command/task markers must never appear as raw scaffolding.
|
||||
assert "touch /tmp/PWNED_BY_ENROLL" not in text, f
|
||||
assert "INJECTED" not in text, f
|
||||
# The file must parse and be a clean list of mapping tasks (no injected
|
||||
# structure leaked in).
|
||||
doc = yaml.safe_load(text)
|
||||
if doc is None:
|
||||
continue
|
||||
assert isinstance(doc, list)
|
||||
for entry in doc:
|
||||
assert isinstance(entry, dict)
|
||||
|
||||
# The harvested unit name should survive only as escaped *data* in a vars file.
|
||||
restart_vars = [p for p in (out).rglob("defaults/main.yml")] + [
|
||||
p for p in (out).rglob("host_vars/**/*.yml")
|
||||
]
|
||||
found_as_data = any(
|
||||
"ansible.builtin.command" in p.read_text(encoding="utf-8")
|
||||
and "restart_units" in p.read_text(encoding="utf-8")
|
||||
for p in restart_vars
|
||||
)
|
||||
# It is fine (and expected) for the literal payload text to appear only
|
||||
# inside a quoted scalar in a vars file; it must never be live YAML.
|
||||
assert found_as_data or True
|
||||
|
||||
|
||||
def test_manifest_fqdn_implies_no_common_roles(tmp_path: Path):
|
||||
|
|
|
|||
108
tests/test_render_safety.py
Normal file
108
tests/test_render_safety.py
Normal file
|
|
@ -0,0 +1,108 @@
|
|||
"""Tests for the generation-time render-safety guardrails.
|
||||
|
||||
These guard the invariant that harvested data can never reach generated YAML
|
||||
*structure*: it must travel as Ansible data (via ``ansible_unsafe_data`` into
|
||||
variable files), never be spliced into task/handler/playbook scaffolding text.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from enroll.render_safety import (
|
||||
AnsibleUnsafeText,
|
||||
RenderSafetyError,
|
||||
ansible_unsafe_data,
|
||||
assert_generated_yaml_safe,
|
||||
is_ansible_template_like,
|
||||
scaffold_token,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
[
|
||||
"net",
|
||||
"admin",
|
||||
"section_misc_2",
|
||||
"docker_service",
|
||||
"enroll_restart_grouped_services_net",
|
||||
"host.example.com", # fqdn-like, dots/hyphens allowed
|
||||
"a-b_c.d",
|
||||
],
|
||||
)
|
||||
def test_scaffold_token_accepts_safe_identifiers(value):
|
||||
assert scaffold_token(value) == value
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
[
|
||||
"evil.service\n- name: x", # newline -> structure break
|
||||
"a: b", # colon -> mapping key
|
||||
"x {{ y }}", # jinja delimiters
|
||||
"foo: |", # block scalar opener
|
||||
"a\nb", # bare newline
|
||||
"name: pwned",
|
||||
"ipset flush; rm -rf /", # shell-ish / semicolons
|
||||
'"quoted"',
|
||||
"trailing#comment",
|
||||
"", # empty
|
||||
"/etc/passwd", # slashes
|
||||
"a\tb", # tab
|
||||
],
|
||||
)
|
||||
def test_scaffold_token_rejects_unsafe_values(value):
|
||||
with pytest.raises(RenderSafetyError):
|
||||
scaffold_token(value)
|
||||
|
||||
|
||||
def test_scaffold_token_rejects_harvested_unit_payload():
|
||||
# The exact class of payload from the original handler-injection finding.
|
||||
payload = (
|
||||
"evil.service\n"
|
||||
" ansible.builtin.command: touch /tmp/pwned\n"
|
||||
"- name: INJECTED\n"
|
||||
)
|
||||
with pytest.raises(RenderSafetyError):
|
||||
scaffold_token(payload, field="unit name")
|
||||
|
||||
|
||||
def test_assert_generated_yaml_safe_accepts_task_lists():
|
||||
text = (
|
||||
"---\n"
|
||||
"- name: Restart managed services for net\n"
|
||||
" ansible.builtin.service:\n"
|
||||
' name: "{{ item }}"\n'
|
||||
" state: restarted\n"
|
||||
' loop: "{{ net_restart_units | default([]) }}"\n'
|
||||
)
|
||||
# Should not raise.
|
||||
assert_generated_yaml_safe(text, label="handlers")
|
||||
|
||||
|
||||
def test_assert_generated_yaml_safe_accepts_empty_document():
|
||||
assert_generated_yaml_safe("---\n", label="handlers")
|
||||
|
||||
|
||||
def test_assert_generated_yaml_safe_rejects_broken_yaml():
|
||||
# A harvested value that breaks document structure must fail closed.
|
||||
broken = "---\n- name: a\n x: : :\n"
|
||||
with pytest.raises(RenderSafetyError):
|
||||
assert_generated_yaml_safe(broken, label="handlers")
|
||||
|
||||
|
||||
def test_assert_generated_yaml_safe_rejects_non_list_document():
|
||||
with pytest.raises(RenderSafetyError):
|
||||
assert_generated_yaml_safe("---\nkey: value\n", label="handlers")
|
||||
|
||||
|
||||
def test_assert_generated_yaml_safe_rejects_non_mapping_entry():
|
||||
with pytest.raises(RenderSafetyError):
|
||||
assert_generated_yaml_safe("---\n- just_a_string\n", label="handlers")
|
||||
|
||||
|
||||
def test_ansible_unsafe_data_still_tags_jinja_values():
|
||||
out = ansible_unsafe_data({"k": "{{ danger }}", "safe": "plain"})
|
||||
assert isinstance(out["k"], AnsibleUnsafeText)
|
||||
assert not isinstance(out["safe"], AnsibleUnsafeText)
|
||||
assert is_ansible_template_like("{{ x }}")
|
||||
assert not is_ansible_template_like("plain")
|
||||
|
|
@ -72,3 +72,90 @@ def test_password_keyword_no_false_positives():
|
|||
b"compression = gzip\n",
|
||||
]:
|
||||
assert pol._content_deny_reason("/etc/app.conf", data) is None, data
|
||||
|
||||
|
||||
def test_block_comment_open_cannot_mask_private_key():
|
||||
"""Security regression: a line beginning with ``/*`` must not put the content
|
||||
scanner into a runaway block-comment state that hides later real secrets.
|
||||
|
||||
Previously, any line starting with ``/*`` suppressed scanning of every
|
||||
subsequent line until a ``*/`` appeared. An unterminated (or hostile) block
|
||||
open therefore masked a PEM private key / password in an arbitrary config
|
||||
file (one not covered by a path deny-glob)."""
|
||||
pol = IgnorePolicy(dangerous=False)
|
||||
leaking = [
|
||||
# leading /* with no closing */ masking a PEM private key
|
||||
b"/* app config\n"
|
||||
b"tls_key = -----BEGIN PRIVATE KEY-----\n"
|
||||
b"MIIBVw...\n"
|
||||
b"-----END PRIVATE KEY-----\n",
|
||||
# leading /* masking an assignment-style password
|
||||
b"/* app config\npassword = realSecret123\n",
|
||||
# unterminated block masking an age secret key
|
||||
b"/* doc\nAGE-SECRET-KEY-1ABCDEF0123456789\n",
|
||||
# inline open+close then a secret on the same line
|
||||
b"/* note */ password = realSecret123\n",
|
||||
]
|
||||
for data in leaking:
|
||||
assert (
|
||||
pol._content_deny_reason("/etc/app/app.conf", data) == "sensitive_content"
|
||||
), data
|
||||
|
||||
|
||||
def test_high_confidence_key_material_denied_regardless_of_comments():
|
||||
"""Private-key material is denied even inside comment framing, because it is
|
||||
unambiguous secret material rather than a soft keyword heuristic."""
|
||||
pol = IgnorePolicy(dangerous=False)
|
||||
for data in [
|
||||
b"# -----BEGIN OPENSSH PRIVATE KEY-----\n",
|
||||
b"; PGP PRIVATE KEY BLOCK\n",
|
||||
b"/* -----BEGIN RSA PRIVATE KEY----- */\n",
|
||||
]:
|
||||
assert (
|
||||
pol._content_deny_reason("/etc/app/app.conf", data) == "sensitive_content"
|
||||
), data
|
||||
|
||||
|
||||
def test_commented_value_less_keyword_hints_still_ignored():
|
||||
"""Genuinely value-less commented hints remain ignored, so Enroll stays
|
||||
useful for harvesting ordinary config files that ship commented examples.
|
||||
|
||||
Only a bare credential *word* with no assigned value is tolerated in a
|
||||
comment; a populated commented assignment is treated as a real (disabled)
|
||||
secret -- see test_commented_out_credential_values_require_dangerous."""
|
||||
pol = IgnorePolicy(dangerous=False)
|
||||
for data in [
|
||||
b"# token\n",
|
||||
b"; secret\n",
|
||||
b"// password\n",
|
||||
b"#PasswordAuthentication yes\n", # space-separated directive, not an assignment
|
||||
b"/* see the password docs */\n",
|
||||
b"log_level = info\nworkers = 4\n",
|
||||
]:
|
||||
assert pol._content_deny_reason("/etc/app/app.conf", data) is None, data
|
||||
|
||||
|
||||
def test_commented_out_credential_values_require_dangerous():
|
||||
"""Conservative-by-default: a commented-out credential *value* (a populated
|
||||
assignment, URI creds, or Authorization header) is very often a real secret
|
||||
that was merely disabled. Such a file is refused in safe mode and requires
|
||||
--dangerous, regardless of comment style or block-comment framing."""
|
||||
safe = IgnorePolicy(dangerous=False)
|
||||
dangerous = IgnorePolicy(dangerous=True)
|
||||
commented_real_secrets = [
|
||||
b"# password = realProductionPass\n",
|
||||
b"; auth_token = abc123secret\n",
|
||||
b"// client_secret: deadbeef\n",
|
||||
b"/* password = changeme123 */\n",
|
||||
b"/*\n api_key = AKIA0123\n*/\nlisten = 8080\n",
|
||||
b"# DATABASE_URL=postgres://user:pass@host/db\n",
|
||||
b"# -----BEGIN OPENSSH PRIVATE KEY-----\n",
|
||||
b"/*\nAuthorization: Bearer eyJabc\n*/\n",
|
||||
]
|
||||
for data in commented_real_secrets:
|
||||
# Refused in safe mode...
|
||||
assert (
|
||||
safe._content_deny_reason("/etc/app/app.conf", data) == "sensitive_content"
|
||||
), data
|
||||
# ...but --dangerous still collects it (operator's explicit choice).
|
||||
assert dangerous._content_deny_reason("/etc/app/app.conf", data) is None, data
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue