hardlinked source files are refused, remote schema URLs require an explicit flag, generated harvest artifacts use the hardened bundle writer, and task/handler/playbook YAML writes go through one safety gate.

tests/test_harvest_safety.py

@@ -320,3 +320,49 @@ def test_ensure_private_empty_dir_creates_private_dir(tmp_path: Path):
     out = hs.ensure_private_empty_dir(tmp_path / "new-cache", label="cache")
     assert out.is_dir()
     assert (out.stat().st_mode & 0o777) == 0o700
+
+
+def test_capture_file_excludes_hardlinked_source(tmp_path: Path):
+    source = tmp_path / "source.conf"
+    source.write_text("safe=true\n", encoding="utf-8")
+    alias = tmp_path / "alias.conf"
+    os.link(source, alias)
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+
+    managed: list[ManagedFile] = []
+    excluded: list[ExcludedFile] = []
+    ok = capture_file(
+        bundle_dir=str(bundle),
+        role_name="role",
+        abs_path=str(alias),
+        reason="test",
+        policy=IgnorePolicy(),
+        path_filter=PathFilter(),
+        managed_out=managed,
+        excluded_out=excluded,
+    )
+
+    assert ok is False
+    assert managed == []
+    assert excluded == [ExcludedFile(path=str(alias), reason="hardlink_source")]
+    assert not (bundle / "artifacts" / "role" / str(alias).lstrip("/")).exists()
+
+
+def test_generated_artifact_writer_refuses_existing_symlink(tmp_path: Path):
+    from enroll.harvest import _write_generated_artifact
+
+    bundle = tmp_path / "bundle"
+    dst_dir = bundle / "artifacts" / "firewall_runtime" / "runtime"
+    dst_dir.mkdir(parents=True)
+    target = tmp_path / "target"
+    target.write_text("old\n", encoding="utf-8")
+    link = dst_dir / "iptables-v4.save"
+    os.symlink(target, link)
+
+    with pytest.raises(OSError):
+        _write_generated_artifact(
+            str(bundle), "firewall_runtime", "runtime/iptables-v4.save", "new\n"
+        )
+
+    assert target.read_text(encoding="utf-8") == "old\n"

tests/test_ignore.py

@@ -416,3 +416,29 @@ def test_inspect_file_still_captures_normal_nested_file(tmp_path: Path):
     assert reason is None
     assert inspection is not None
     assert inspection.data == b"workers=4\n"
+
+
+def test_inspect_file_refuses_hardlinked_source(tmp_path: Path):
+    pol = IgnorePolicy()
+    original = tmp_path / "original.conf"
+    original.write_text("safe=true\n", encoding="utf-8")
+    alias = tmp_path / "alias.conf"
+    os.link(original, alias)
+
+    reason, inspection = pol.inspect_file(str(alias))
+
+    assert reason == "hardlink_source"
+    assert inspection is None
+
+
+def test_inspect_file_refuses_hardlinked_source_even_in_dangerous_mode(tmp_path: Path):
+    pol = IgnorePolicy(dangerous=True)
+    original = tmp_path / "original.conf"
+    original.write_text("safe=true\n", encoding="utf-8")
+    alias = tmp_path / "alias.conf"
+    os.link(original, alias)
+
+    reason, inspection = pol.inspect_file(str(alias))
+
+    assert reason == "hardlink_source"
+    assert inspection is None

tests/test_render_safety.py

@@ -106,3 +106,16 @@ def test_ansible_unsafe_data_still_tags_jinja_values():
     assert not isinstance(out["safe"], AnsibleUnsafeText)
     assert is_ansible_template_like("{{ x }}")
     assert not is_ansible_template_like("plain")
+
+
+def test_write_generated_task_yaml_uses_safety_gate(tmp_path):
+    from enroll.ansible import _write_generated_task_yaml
+
+    path = tmp_path / "tasks.yml"
+
+    with pytest.raises(RenderSafetyError):
+        _write_generated_task_yaml(
+            str(path), "---\nkey: value\n", label="test tasks/main.yml"
+        )
+
+    assert not path.exists()

tests/test_validate.py

@@ -305,16 +305,54 @@ def test_validate_harvest_invalid_json(tmp_path: Path):
     assert any("failed to parse" in e for e in result.errors)
 
 
-def test_validate_harvest_schema_error(tmp_path: Path):
+def test_validate_harvest_rejects_remote_schema_by_default(tmp_path: Path):
     bundle_dir = tmp_path / "bundle"
     bundle_dir.mkdir()
     state_file = bundle_dir / "state.json"
     state_file.write_text("{}", encoding="utf-8")
+
     result = validate_harvest(
         str(bundle_dir), schema="https://invalid.invalid/schema.json"
     )
+
     assert result.ok is False
-    assert any("failed to load/validate schema" in e for e in result.errors)
+    assert any("remote schema URLs are disabled" in e for e in result.errors)
+
+
+def test_validate_harvest_allows_remote_schema_when_explicit(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+):
+    bundle_dir = tmp_path / "bundle"
+    bundle_dir.mkdir()
+    state_file = bundle_dir / "state.json"
+    state_file.write_text("{}", encoding="utf-8")
+
+    class _Response:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def read(self):
+            return b'{"type":"object"}'
+
+    called = []
+
+    def _fake_urlopen(url: str, timeout: int):
+        called.append((url, timeout))
+        return _Response()
+
+    monkeypatch.setattr("enroll.validate.urllib.request.urlopen", _fake_urlopen)
+
+    result = validate_harvest(
+        str(bundle_dir),
+        schema="https://example.invalid/schema.json",
+        allow_remote_schema=True,
+    )
+
+    assert called == [("https://example.invalid/schema.json", 10)]
+    assert not any("remote schema URLs are disabled" in e for e in result.errors)
 
 
 def test_validate_harvest_missing_artifact(tmp_path: Path):