Public Access
368 lines
14 KiB
Python
368 lines
14 KiB
Python
"""Unit tests for devx.tools.check_ansible_no_log."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from click.testing import CliRunner
|
|
|
|
from devx.tools.check_ansible_no_log import _check_task, check_directory, main
|
|
|
|
|
|
def _make_task(name: str, action: str, value: str, **extra: object) -> dict:
|
|
"""Build a minimal task dict for testing."""
|
|
task: dict = {"name": name, action: value}
|
|
task.update(extra)
|
|
return task
|
|
|
|
|
|
class TestCheckTask:
|
|
def test_task_with_secret_and_no_log_passes(self):
|
|
task = _make_task(
|
|
"Safe task",
|
|
"ansible.builtin.shell",
|
|
"echo {{ _secrets.mattermost_admin_password }}",
|
|
no_log=True,
|
|
)
|
|
assert _check_task(task, Path("test.yml"), 1) == []
|
|
|
|
def test_task_with_secret_and_no_no_log_fails(self):
|
|
task = _make_task(
|
|
"Unsafe task",
|
|
"ansible.builtin.shell",
|
|
"echo {{ _secrets.mattermost_admin_password }}",
|
|
)
|
|
violations = _check_task(task, Path("test.yml"), 1)
|
|
assert len(violations) == 1
|
|
assert "no_log" in violations[0]
|
|
|
|
def test_task_without_secret_passes(self):
|
|
task = _make_task(
|
|
"Normal task",
|
|
"ansible.builtin.shell",
|
|
"echo hello world",
|
|
)
|
|
assert _check_task(task, Path("test.yml"), 1) == []
|
|
|
|
def test_task_with_jinja_no_log_passes(self):
|
|
task = _make_task(
|
|
"Safe task with jinja no_log",
|
|
"ansible.builtin.shell",
|
|
"echo {{ _secrets.mattermost_admin_password }}",
|
|
no_log="{{ not (debug_mode | default(false) | bool) }}",
|
|
)
|
|
assert _check_task(task, Path("test.yml"), 1) == []
|
|
|
|
def test_task_with_password_in_name_only_no_false_positive(self):
|
|
"""Task name contains 'password' but no secret value — should not flag."""
|
|
task = _make_task(
|
|
"Configure passwdqc in common-password",
|
|
"ansible.builtin.lineinfile",
|
|
"password required pam_passwdqc.so min=disabled,disabled,16,12,8",
|
|
)
|
|
assert _check_task(task, Path("test.yml"), 1) == []
|
|
|
|
def test_task_with_password_in_module_param_no_false_positive(self):
|
|
"""Module param named 'password' but value is a literal — no Jinja."""
|
|
task = {
|
|
"name": "Set user password",
|
|
"ansible.builtin.user": {
|
|
"name": "deploy",
|
|
"password_lock": True,
|
|
},
|
|
}
|
|
assert _check_task(task, Path("test.yml"), 1) == []
|
|
|
|
def test_task_with_vault_password_variable_fails(self):
|
|
task = _make_task(
|
|
"Unsafe vault task",
|
|
"ansible.builtin.shell",
|
|
"echo {{ vault_zitadel_db_password }}",
|
|
)
|
|
violations = _check_task(task, Path("test.yml"), 1)
|
|
assert len(violations) == 1
|
|
|
|
def test_task_with_nested_dict_secret_fails(self):
|
|
"""Secrets in nested dict values (e.g. set_fact) should be caught."""
|
|
task = {
|
|
"name": "Set secrets",
|
|
"ansible.builtin.set_fact": {
|
|
"db_password": "{{ vault_db_password }}",
|
|
"api_key": "{{ vault_api_key }}",
|
|
},
|
|
}
|
|
violations = _check_task(task, Path("test.yml"), 1)
|
|
assert len(violations) == 1
|
|
|
|
def test_task_with_no_log_none_passes(self):
|
|
"""no_log: None should count as not set (flagged)."""
|
|
task = _make_task(
|
|
"Unsafe task",
|
|
"ansible.builtin.shell",
|
|
"echo {{ _secrets.db_password }}",
|
|
no_log=None,
|
|
)
|
|
violations = _check_task(task, Path("test.yml"), 1)
|
|
assert len(violations) == 1
|
|
|
|
def test_task_with_secret_in_list_value_fails(self):
|
|
"""Secrets inside list values should be caught."""
|
|
task = {
|
|
"name": "Task with list secret",
|
|
"ansible.builtin.set_fact": {
|
|
"items": ["{{ _secrets.api_key }}", "normal_value"],
|
|
},
|
|
}
|
|
violations = _check_task(task, Path("test.yml"), 1)
|
|
assert len(violations) == 1
|
|
|
|
def test_task_with_api_key_secret_fails(self):
|
|
"""api_key in Jinja expression should be caught."""
|
|
task = _make_task(
|
|
"Unsafe task",
|
|
"ansible.builtin.shell",
|
|
"echo {{ my_api_key }}",
|
|
)
|
|
violations = _check_task(task, Path("test.yml"), 1)
|
|
assert len(violations) == 1
|
|
|
|
def test_task_with_secret_in_jinja_fails(self):
|
|
"""_secret in Jinja expression should be caught."""
|
|
task = _make_task(
|
|
"Unsafe task",
|
|
"ansible.builtin.shell",
|
|
"echo {{ my_secret }}",
|
|
)
|
|
violations = _check_task(task, Path("test.yml"), 1)
|
|
assert len(violations) == 1
|
|
|
|
def test_task_with_access_token_fails(self):
|
|
"""access_token in Jinja expression should be caught."""
|
|
task = _make_task(
|
|
"Unsafe task",
|
|
"ansible.builtin.shell",
|
|
"echo {{ my_access_token }}",
|
|
)
|
|
violations = _check_task(task, Path("test.yml"), 1)
|
|
assert len(violations) == 1
|
|
|
|
def test_task_with_non_secret_non_dict_non_list_value(self):
|
|
"""Non-str, non-dict, non-list values (e.g. int) should not crash."""
|
|
task = _make_task(
|
|
"Task with int",
|
|
"ansible.builtin.shell",
|
|
"echo hello",
|
|
some_int=42,
|
|
)
|
|
assert _check_task(task, Path("test.yml"), 1) == []
|
|
|
|
|
|
class TestCheckDirectory:
|
|
def test_clean_directory_passes(self, tmp_path: Path):
|
|
"""A directory with no secret-handling tasks should pass."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text(
|
|
"- name: Normal task\n ansible.builtin.shell: echo hello\n changed_when: false\n"
|
|
)
|
|
assert check_directory(role_dir) == []
|
|
|
|
def test_unsafe_task_is_caught(self, tmp_path: Path):
|
|
"""A task with secrets but no no_log should be flagged."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text(
|
|
"- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n changed_when: false\n"
|
|
)
|
|
violations = check_directory(role_dir)
|
|
assert len(violations) == 1
|
|
assert "Unsafe task" in violations[0]
|
|
|
|
def test_molecule_files_are_skipped(self, tmp_path: Path):
|
|
"""Molecule test files should not be scanned."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
mol_dir = role_dir / "molecule" / "default" / "tasks"
|
|
mol_dir.mkdir(parents=True)
|
|
(mol_dir / "main.yml").write_text(
|
|
"- name: Unsafe task in molecule\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
|
|
)
|
|
assert check_directory(role_dir) == []
|
|
|
|
def test_playbook_format_is_parsed(self, tmp_path: Path):
|
|
"""Playbook files (list of plays with 'hosts') should be parsed."""
|
|
pb_dir = tmp_path / "playbooks"
|
|
pb_dir.mkdir(parents=True)
|
|
(pb_dir / "test.yml").write_text(
|
|
"---\n"
|
|
"- name: Test play\n"
|
|
" hosts: all\n"
|
|
" tasks:\n"
|
|
" - name: Unsafe task\n"
|
|
" ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
|
|
)
|
|
violations = check_directory(tmp_path)
|
|
assert len(violations) == 1
|
|
assert "Unsafe task" in violations[0]
|
|
|
|
def test_invalid_yaml_is_skipped(self, tmp_path: Path):
|
|
"""Invalid YAML files should be skipped, not crash."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text("{{ invalid yaml: [")
|
|
assert check_directory(role_dir) == []
|
|
|
|
def test_empty_yaml_doc_is_skipped(self, tmp_path: Path):
|
|
"""Empty YAML documents (None) should be skipped."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text("---\n")
|
|
assert check_directory(role_dir) == []
|
|
|
|
def test_non_dict_non_list_doc_is_skipped(self, tmp_path: Path):
|
|
"""YAML docs that are neither dict nor list should be skipped."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text("just a string\n")
|
|
assert check_directory(role_dir) == []
|
|
|
|
def test_task_file_with_non_dict_task_skipped(self, tmp_path: Path):
|
|
"""Non-dict items in a task list should be skipped."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text(
|
|
"- just a string\n- name: Safe task\n ansible.builtin.shell: echo hello\n"
|
|
)
|
|
assert check_directory(role_dir) == []
|
|
|
|
def test_secret_in_list_value_is_caught(self, tmp_path: Path):
|
|
"""Secrets inside list values should be caught."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text(
|
|
"- name: Task with list secret\n"
|
|
" ansible.builtin.set_fact:\n"
|
|
" items:\n"
|
|
' - "{{ _secrets.api_key }}"\n'
|
|
" - normal_value\n"
|
|
)
|
|
violations = check_directory(role_dir)
|
|
assert len(violations) == 1
|
|
|
|
def test_single_play_dict_format(self, tmp_path: Path):
|
|
"""A playbook that's a bare dict (not list of plays) should be parsed."""
|
|
pb_dir = tmp_path / "playbooks"
|
|
pb_dir.mkdir(parents=True)
|
|
(pb_dir / "test.yml").write_text(
|
|
"---\n"
|
|
"name: Single play\n"
|
|
"hosts: all\n"
|
|
"tasks:\n"
|
|
" - name: Unsafe task\n"
|
|
" ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
|
|
)
|
|
violations = check_directory(tmp_path)
|
|
assert len(violations) == 1
|
|
|
|
def test_play_with_non_dict_play_skipped(self, tmp_path: Path):
|
|
"""Non-dict plays in a playbook list should be skipped."""
|
|
pb_dir = tmp_path / "playbooks"
|
|
pb_dir.mkdir(parents=True)
|
|
# First play is valid (makes is_plays=True), second is a non-dict
|
|
(pb_dir / "test.yml").write_text(
|
|
"---\n"
|
|
"- name: Safe play\n"
|
|
" hosts: all\n"
|
|
" tasks:\n"
|
|
" - name: Safe task\n"
|
|
" ansible.builtin.shell: echo hello\n"
|
|
'- "just a string as second play"\n'
|
|
)
|
|
assert check_directory(tmp_path) == []
|
|
|
|
def test_play_with_non_list_tasks_skipped(self, tmp_path: Path):
|
|
"""Plays where tasks is not a list should be skipped."""
|
|
pb_dir = tmp_path / "playbooks"
|
|
pb_dir.mkdir(parents=True)
|
|
(pb_dir / "test.yml").write_text('---\n- name: Play with bad tasks\n hosts: all\n tasks: "not a list"\n')
|
|
assert check_directory(tmp_path) == []
|
|
|
|
def test_play_with_non_dict_task_in_playbook(self, tmp_path: Path):
|
|
"""Non-dict tasks in a playbook should be skipped."""
|
|
pb_dir = tmp_path / "playbooks"
|
|
pb_dir.mkdir(parents=True)
|
|
(pb_dir / "test.yml").write_text(
|
|
"---\n"
|
|
"- name: Play\n"
|
|
" hosts: all\n"
|
|
" tasks:\n"
|
|
' - "just a string"\n'
|
|
" - name: Safe task\n"
|
|
" ansible.builtin.shell: echo hello\n"
|
|
)
|
|
assert check_directory(tmp_path) == []
|
|
|
|
def test_yaml_file_with_oserror_skipped(self, tmp_path: Path):
|
|
"""YAML files that can't be opened should be skipped."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
# Create a file that will cause OSError when opened
|
|
# (use a directory with .yml extension)
|
|
bad_file = role_dir / "tasks" / "main.yml"
|
|
bad_file.mkdir()
|
|
assert check_directory(role_dir) == []
|
|
|
|
|
|
class TestMain:
|
|
def test_main_passes_on_clean_dir(self, tmp_path: Path):
|
|
"""main() should exit 0 on a clean directory."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text(
|
|
"- name: Normal task\n ansible.builtin.shell: echo hello\n changed_when: false\n"
|
|
)
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--path", str(role_dir)])
|
|
assert result.exit_code == 0
|
|
assert "OK" in result.output or "no_log" in result.output
|
|
|
|
def test_main_fails_on_unsafe_dir(self, tmp_path: Path):
|
|
"""main() should exit 1 when violations are found."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text(
|
|
"- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
|
|
)
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--path", str(role_dir)])
|
|
assert result.exit_code == 1
|
|
assert "Unsafe task" in result.output
|
|
|
|
def test_main_returns_2_on_missing_dir(self, tmp_path: Path):
|
|
"""main() should exit 2 when the directory doesn't exist."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--path", str(tmp_path / "nonexistent")])
|
|
assert result.exit_code == 2
|
|
|
|
def test_main_with_ansible_dir_option(self, tmp_path: Path):
|
|
"""main() --ansible-dir should work like --path."""
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text(
|
|
"- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
|
|
)
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--ansible-dir", str(tmp_path)])
|
|
assert result.exit_code == 1
|
|
|
|
def test_main_no_path_no_ansible_dir_uses_default(self, tmp_path: Path, monkeypatch):
|
|
"""main() with no args uses DEFAULT_ANSIBLE_DIR."""
|
|
import devx.tools.check_ansible_no_log as mod
|
|
|
|
role_dir = tmp_path / "roles" / "test_role"
|
|
(role_dir / "tasks").mkdir(parents=True)
|
|
(role_dir / "tasks" / "main.yml").write_text("- name: Normal task\n ansible.builtin.shell: echo hello\n")
|
|
monkeypatch.setattr(mod, "DEFAULT_ANSIBLE_DIR", tmp_path)
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, [])
|
|
assert result.exit_code == 0
|