| import io |
| import json |
| import tarfile |
|
|
| import pytest |
|
|
| from core.errors import ErrorCode, MatterOSError |
| from core.io import MAX_INPUT_BYTES, loads |
| from storage.archive import read_archive |
| from storage.artifacts import safe_join |
|
|
|
|
| def test_path_traversal_blocked(tmp_path): |
| assert safe_join(tmp_path, "a/b.txt").parent.parent == tmp_path.resolve() |
| for bad in ("../x", "/etc/passwd", "a/../../x"): |
| with pytest.raises(MatterOSError): |
| safe_join(tmp_path, bad) |
|
|
|
|
| def test_artifact_names_cannot_escape(app): |
| a = app.artifacts.put_bytes(b"x", "test", "../../evil.txt") |
| assert ".." not in a["path"] and a["path"].endswith("evil.txt") |
|
|
|
|
| def test_unsafe_archive_members_rejected(): |
| buf = io.BytesIO() |
| with tarfile.open(fileobj=buf, mode="w:gz") as tf: |
| d = b"x"; ti = tarfile.TarInfo("../escape.txt"); ti.size = 1; tf.addfile(ti, io.BytesIO(d)) |
| m = json.dumps({"members": {}}).encode(); ti = tarfile.TarInfo("manifest.json"); ti.size = len(m); tf.addfile(ti, io.BytesIO(m)) |
| with pytest.raises(MatterOSError): |
| read_archive(buf.getvalue()) |
|
|
|
|
| def test_parser_limits_and_untrusted_input(): |
| with pytest.raises(MatterOSError) as e: |
| loads("1\n\nH 0 0 0\n" + " " * (MAX_INPUT_BYTES + 1), "xyz") |
| assert e.value.code == ErrorCode.RESOURCE_LIMIT |
| with pytest.raises(MatterOSError): |
| loads("__import__('os').system('echo pwned')", "xyz") |
|
|
|
|
| def test_secrets_redacted_from_audit_logs(app): |
| app.db.audit("tester", "probe", "x", {"note": "api_key=SUPERSECRET123 password: hunter2"}) |
| row = app.db.one("SELECT details FROM audit_log WHERE action='probe' ORDER BY id DESC LIMIT 1") |
| assert "SUPERSECRET123" not in row["details"] and "hunter2" not in row["details"] |
|
|
|
|
| def test_no_shell_tool_and_plugin_allowlist(): |
| from agents.tools import REGISTRY |
| assert not any(k.startswith(("shell", "os.", "fs.", "exec")) for k in REGISTRY) |
| from simulation.registry import _load_adapter |
| with pytest.raises(ValueError): |
| _load_adapter("os:system", None) |
|
|
|
|
| def test_paid_services_disabled_by_default(app): |
| assert app.config["llm"]["paid_services_enabled"] is False |
| assert app.policy["paid_services"]["enabled"] is False and app.policy["cloud_workers"]["enabled"] is False |
|
|
|
|
| def test_worker_isolation_cpu_limit_and_cancel_flag_scoped(app, project): |
| """Jobs run in a separate process with their own working directory under the jobs dir.""" |
| from discovery.generators import lj_random_cluster |
| c = app.candidates.register(lj_random_cluster(4, 1), project.project_id, "t") |
| rec = app.simulation.submit_run(c.candidate_id, "ase_lj", "energy", {}) |
| j = app.jobs.get(rec.provenance["job_id"]) |
| assert str(app.jobs.jobs_dir) in j["checkpoint_path"] |
|
|
