yingqi-z20-Agent-libOS

#1

from __future__ import annotations

#2

#3

import unittest

#4

from uuid import uuid4

#5

#6

from agent_libos import Runtime

#7

from agent_libos.models import CapabilityRight, ForkMode, HumanRequestStatus

#8

#9

#10

class ExternalBoundaryTests(unittest.TestCase):

#11

def setUp(self) -> None:

#12

self.runtime = Runtime.open("local")

#13

self.human_output: list[str] = []

#14

self.runtime.substrate.human.output_sink = self.human_output.append

#15

#16

def tearDown(self) -> None:

#17

self.runtime.close()

#18

#19

def test_read_file_tool_cannot_bypass_filesystem_capability(self) -> None:

#20

path = self._write_workspace_fixture("hello from workspace")

#21

pid = self.runtime.process.spawn(image="review-agent:v0", goal="read a file")

#22

#23

denied = self.runtime.tools.call(pid, "read_text_file", {"path": path})

#24

self.assertFalse(denied.ok)

#25

self.assertIn("lacks read", denied.error or "")

#26

self.assertNotIn("primitive.filesystem.read_text", self._audit_actions())

#27

#28

self.runtime.filesystem.grant_path(pid, path, [CapabilityRight.READ], issued_by="test")

#29

allowed = self.runtime.tools.call(pid, "read_text_file", {"path": path})

#30

self.assertTrue(allowed.ok)

#31

self.assertEqual(allowed.payload["content"], "hello from workspace")

#32

self.assertIn("primitive.filesystem.read_text", self._audit_actions())

#33

#34

def test_write_file_tool_cannot_bypass_filesystem_capability(self) -> None:

#35

path = f"agent_outputs/boundary_write_{uuid4().hex}.txt"

#36

target = self.runtime.workspace_root / path

#37

pid = self.runtime.process.spawn(image="review-agent:v0", goal="write a file")

#38

#39

denied = self.runtime.tools.call(pid, "write_text_file", {"path": path, "content": "denied"})

#40

self.assertFalse(denied.ok)

#41

self.assertFalse(target.exists())

#42

#43

self.runtime.filesystem.grant_path(pid, path, [CapabilityRight.WRITE], issued_by="test")

#44

allowed = self.runtime.tools.call(pid, "write_text_file", {"path": path, "content": "allowed"})

#45

self.assertTrue(allowed.ok)

#46

self.assertEqual(target.read_text(encoding="utf-8"), "allowed")

#47

self.assertIn("primitive.filesystem.write_text", self._audit_actions())

#48

#49

def test_write_precondition_does_not_leak_existing_file_without_capability(self) -> None:

#50

path = self._write_workspace_fixture("existing")

#51

pid = self.runtime.process.spawn(image="review-agent:v0", goal="probe existing file")

#52

#53

denied = self.runtime.tools.call(

#54

pid,

#55

"write_text_file",

#56

{"path": path, "content": "new", "overwrite": False},

#57

)

#58

#59

self.assertFalse(denied.ok)

#60

self.assertIn("lacks write", denied.error or "")

#61

self.assertNotIn("already exists", denied.error or "")

#62

self.assertEqual((self.runtime.workspace_root / path).read_text(encoding="utf-8"), "existing")

#63

#64

def test_delete_precondition_does_not_leak_missing_file_without_capability(self) -> None:

#65

path = f"agent_outputs/missing_delete_{uuid4().hex}.txt"

#66

pid = self.runtime.process.spawn(image="review-agent:v0", goal="probe missing file")

#67

#68

denied = self.runtime.tools.call(pid, "delete_file", {"path": path})

#69

#70

self.assertFalse(denied.ok)

#71

self.assertIn("lacks delete", denied.error or "")

#72

self.assertNotIn("does not exist", denied.error or "")

#73

#74

def test_human_output_tool_cannot_bypass_human_capability(self) -> None:

#75

pid = self.runtime.process.spawn(image="review-agent:v0", goal="speak to the human")

#76

#77

denied = self.runtime.tools.call(pid, "human_output", {"message": "denied"})

#78

self.assertFalse(denied.ok)

#79

self.assertEqual(self.human_output, [])

#80

#81

self.runtime.capability.grant(pid, "human:owner", [CapabilityRight.WRITE], issued_by="test")

#82

allowed = self.runtime.tools.call(pid, "human_output", {"message": "allowed"})

#83

self.assertTrue(allowed.ok)

#84

self.assertEqual(self.human_output, ["allowed"])

#85

self.assertEqual(self.runtime.human.list(pid)[0].status, HumanRequestStatus.DELIVERED)

#86

self.assertIn("human.output", self._audit_actions())

#87

#88

def test_process_cannot_call_tool_outside_creation_tool_table(self) -> None:

#89

pid = self.runtime.process.spawn(image="toolmaker-agent:v0", goal="call unavailable tool")

#90

#91

denied = self.runtime.tools.call(pid, "write_text_file", {"path": "agent_outputs/no_tool.txt", "content": "x"})

#92

#93

self.assertFalse(denied.ok)

#94

self.assertIn("not in process tool table", denied.error or "")

#95

self.assertNotIn("human.query", self._audit_actions())

#96

#97

def test_path_escape_is_denied_by_filesystem_primitive(self) -> None:

#98

pid = self.runtime.process.spawn(image="review-agent:v0", goal="escape workspace")

#99

self.runtime.filesystem.grant_workspace(pid, [CapabilityRight.WRITE], issued_by="test")

#100

#101

denied = self.runtime.tools.call(pid, "write_text_file", {"path": "../outside.txt", "content": "denied"})

#102

#103

self.assertFalse(denied.ok)

#104

self.assertIn("escapes filesystem adapter root", denied.error or "")

#105

self.assertNotIn("primitive.filesystem.write_text", self._audit_actions())

#106

#107

def test_revoked_filesystem_capability_denies_write(self) -> None:

#108

path = f"agent_outputs/revoked_write_{uuid4().hex}.txt"

#109

pid = self.runtime.process.spawn(image="review-agent:v0", goal="revoked write")

#110

cap = self.runtime.filesystem.grant_path(pid, path, [CapabilityRight.WRITE], issued_by="test")

#111

self.runtime.capability.revoke(cap.cap_id, revoked_by="test", reason="boundary test")

#112

#113

denied = self.runtime.tools.call(pid, "write_text_file", {"path": path, "content": "denied"})

#114

#115

self.assertFalse(denied.ok)

#116

self.assertFalse((self.runtime.workspace_root / path).exists())

#117

self.assertNotIn("primitive.filesystem.write_text", self._audit_actions())

#118

#119

def test_fork_does_not_inherit_parent_filesystem_write_capability(self) -> None:

#120

path = f"agent_outputs/fork_write_{uuid4().hex}.txt"

#121

parent = self.runtime.process.spawn(image="review-agent:v0", goal="parent")

#122

self.runtime.filesystem.grant_path(parent, path, [CapabilityRight.WRITE], issued_by="test")

#123

child = self.runtime.process.fork(parent, goal="child", mode=ForkMode.WORKER)

#124

#125

denied = self.runtime.tools.call(child, "write_text_file", {"path": path, "content": "denied"})

#126

allowed = self.runtime.tools.call(parent, "write_text_file", {"path": path, "content": "allowed"})

#127

#128

self.assertFalse(denied.ok)

#129

self.assertTrue(allowed.ok)

#130

self.assertEqual((self.runtime.workspace_root / path).read_text(encoding="utf-8"), "allowed")

#131

#132

def _write_workspace_fixture(self, content: str) -> str:

#133

path = f"agent_outputs/boundary_read_{uuid4().hex}.txt"

#134

target = self.runtime.workspace_root / path

#135

target.parent.mkdir(parents=True, exist_ok=True)

#136

target.write_text(content, encoding="utf-8")

#137

return path

#138

#139

def _audit_actions(self) -> list[str]:

#140

return [record.action for record in self.runtime.audit.trace()]

#141

#142

#143

if __name__ == "__main__":

#144

unittest.main()

#145

z6MkqRzA/yingqi-z20-Agent-libOS