yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	import os
#4	import asyncio
#5	import subprocess
#6	from dataclasses import dataclass
#7	from pathlib import PurePath
#8	from typing import Any
#9
#10	from agent_libos.capability.manager import CapabilityManager
#11	from agent_libos.config import DEFAULT_CONFIG, AgentLibOSConfig, ShellCommandRule, ShellPolicyLevel
#12	from agent_libos.models import Capability, CapabilityRight, EventType
#13	from agent_libos.models.exceptions import CapabilityDenied, HumanApprovalRequired, ValidationError
#14	from agent_libos.runtime.audit_manager import AuditManager
#15	from agent_libos.runtime.event_bus import EventBus
#16	from agent_libos.substrate import CommandResult, LocalShellProvider, ShellProvider
#17	from agent_libos.utils.ids import utc_now
#18
#19	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#20
#21	_WINDOWS_EXECUTABLE_SUFFIXES = (".exe", ".cmd", ".bat", ".com", ".ps1")
#22
#23
#24	@dataclass(frozen=True)
#25	class ShellPolicyDecision:
#26	allowed: bool
#27	ask_human: bool
#28	reason: str
#29	policy_level: str \| None
#30	matched_rule: tuple[str, ...] \| None = None
#31	high_risk: bool = False
#32	consume_once: bool = False
#33
#34
#35	class ShellAdapter:
#36	"""Capability-checked shell primitive.
#37
#38	Commands are accepted only as argv arrays and are executed by the substrate
#39	with shell=False. Allow/ask decisions use exact token rules; no glob,
#40	substring, or shell-style parsing is used for whitelist matching.
#41	"""
#42
#43	def __init__(
#44	self,
#45	capabilities: CapabilityManager,
#46	audit: AuditManager,
#47	events: EventBus \| None = None,
#48	cwd: str \| os.PathLike[str] \| None = None,
#49	human: Any \| None = None,
#50	provider: ShellProvider \| None = None,
#51	config: AgentLibOSConfig \| None = None,
#52	):
#53	self.config = config or DEFAULT_CONFIG
#54	self.capabilities = capabilities
#55	self.audit = audit
#56	self.events = events
#57	self.human = human
#58	self.provider = provider or LocalShellProvider(cwd or ".")
#59
#60	def run(
#61	self,
#62	pid: str,
#63	argv: list[str],
#64	timeout: float = _TOOL_DEFAULTS.shell_timeout_s,
#65	cwd: str \| os.PathLike[str] \| None = None,
#66	) -> CommandResult:
#67	checked = self._validate_argv(argv)
#68	selected_timeout = self._validate_timeout(timeout)
#69	resource = self.resource_for(checked)
#70	decision = self._authorize(pid, checked, resource, timeout=selected_timeout)
#71	if decision.ask_human:
#72	self._request_human_approval(pid, checked, resource, decision, timeout=selected_timeout, cwd=cwd)
#73	if not decision.allowed:
#74	raise CapabilityDenied(f"{pid} denied shell execute on {resource}: {decision.reason}")
#75	try:
#76	if cwd is None:
#77	proc = self.provider.run(checked, timeout=selected_timeout)
#78	else:
#79	proc = self.provider.run(checked, timeout=selected_timeout, cwd=os.fspath(cwd))
#80	proc = self._bounded_result(proc)
#81	except subprocess.TimeoutExpired as exc:
#82	self.audit.record(
#83	actor=pid,
#84	action="primitive.shell.timeout",
#85	target=resource,
#86	decision={"argv": checked, "timeout_s": selected_timeout, "cwd": os.fspath(cwd) if cwd is not None else None},
#87	)
#88	raise TimeoutError(f"shell command timed out after {selected_timeout}s: {checked}") from exc
#89	finally:
#90	if decision.consume_once:
#91	self.capabilities.consume_allow_once(
#92	subject=pid,
#93	resource=resource,
#94	right=CapabilityRight.EXECUTE,
#95	used_by="shell",
#96	)
#97	self._emit_run_event(pid, resource, checked, proc, decision, cwd=cwd)
#98	self.audit.record(
#99	actor=pid,
#100	action="primitive.shell.run",
#101	target=resource,
#102	decision={
#103	"argv": checked,
#104	"returncode": proc.returncode,
#105	"policy_level": decision.policy_level,
#106	"policy_reason": decision.reason,
#107	"matched_rule": list(decision.matched_rule) if decision.matched_rule else None,
#108	"high_risk": decision.high_risk,
#109	"cwd": os.fspath(cwd) if cwd is not None else None,
#110	"stdout_truncated": proc.stdout_truncated,
#111	"stderr_truncated": proc.stderr_truncated,
#112	},
#113	)
#114	return proc
#115
#116	async def arun(
#117	self,
#118	pid: str,
#119	argv: list[str],
#120	timeout: float = _TOOL_DEFAULTS.shell_timeout_s,
#121	cwd: str \| os.PathLike[str] \| None = None,
#122	) -> CommandResult:
#123	return await asyncio.to_thread(self.run, pid, argv, timeout=timeout, cwd=cwd)
#124
#125	def grant_policy(
#126	self,
#127	pid: str,
#128	level: ShellPolicyLevel \| str \| None = None,
#129	*,
#130	issued_by: str = "shell",
#131	) -> Capability:
#132	selected = self._normalize_policy_level(level or self.config.shell.default_policy_level)
#133	return self.capabilities.grant(
#134	subject=pid,
#135	resource=self.policy_resource(),
#136	rights=[CapabilityRight.EXECUTE],
#137	issued_by=issued_by,
#138	constraints={self.config.shell.policy_capability_key: selected},
#139	)
#140
#141	def policy_resource(self) -> str:
#142	return self.config.shell.policy_resource
#143
#144	def resource_for(self, argv: list[str]) -> str:
#145	command = self._command_identity(argv[0])
#146	return f"shell:{command}"
#147
#148	def _authorize(self, pid: str, argv: list[str], resource: str, *, timeout: float) -> ShellPolicyDecision:
#149	policy_caps = self._matching_shell_policy_caps(pid, resource)
#150	if any(
#151	cap.constraints.get(self.config.shell.policy_capability_key) == self.config.shell.always_deny_level
#152	for cap in policy_caps
#153	):
#154	return ShellPolicyDecision(
#155	allowed=False,
#156	ask_human=False,
#157	reason="shell policy is always_deny",
#158	policy_level=self.config.shell.always_deny_level,
#159	)
#160
#161	legacy_policy, legacy_consume_once = self._legacy_permission_policy(pid, resource)
#162	if legacy_policy == CapabilityManager.ALWAYS_DENY:
#163	return ShellPolicyDecision(False, False, "permission policy denied command", legacy_policy)
#164	if legacy_policy == CapabilityManager.ASK_EACH_TIME:
#165	return ShellPolicyDecision(False, True, "permission policy requires approval", legacy_policy)
#166	if legacy_policy in {CapabilityManager.ALWAYS_ALLOW, CapabilityManager.ALLOW_ONCE}:
#167	return ShellPolicyDecision(
#168	allowed=True,
#169	ask_human=False,
#170	reason="explicit command capability allowed command",
#171	policy_level=legacy_policy,
#172	consume_once=legacy_consume_once,
#173	)
#174
#175	if not policy_caps:
#176	raise CapabilityDenied(f"{pid} lacks shell execute policy for {resource}")
#177
#178	level = self._selected_policy_level(policy_caps)
#179	whitelist_match = self._first_matching_rule(argv, self.config.shell.whitelist, allow_bare_only=True)
#180	blacklist_match = self._first_matching_blacklist_rule(argv)
#181
#182	if level == self.config.shell.allowlist_auto_else_ask_level:
#183	if whitelist_match is not None:
#184	return ShellPolicyDecision(
#185	allowed=True,
#186	ask_human=False,
#187	reason="command matched shell whitelist",
#188	policy_level=level,
#189	matched_rule=whitelist_match.argv,
#190	)
#191	return ShellPolicyDecision(
#192	allowed=False,
#193	ask_human=True,
#194	reason="command did not match shell whitelist",
#195	policy_level=level,
#196	)
#197	if level == self.config.shell.blocklist_ask_else_auto_level:
#198	if blacklist_match is not None:
#199	return ShellPolicyDecision(
#200	allowed=False,
#201	ask_human=True,
#202	reason="command matched shell blacklist",
#203	policy_level=level,
#204	matched_rule=blacklist_match.argv,
#205	)
#206	return ShellPolicyDecision(
#207	allowed=True,
#208	ask_human=False,
#209	reason="command did not match shell blacklist",
#210	policy_level=level,
#211	)
#212	if level == self.config.shell.always_allow_level:
#213	return ShellPolicyDecision(
#214	allowed=True,
#215	ask_human=False,
#216	reason="shell policy is always_allow",
#217	policy_level=level,
#218	high_risk=True,
#219	)
#220	return ShellPolicyDecision(False, False, f"unsupported shell policy level: {level}", level)
#221
#222	def _request_human_approval(
#223	self,
#224	pid: str,
#225	argv: list[str],
#226	resource: str,
#227	decision: ShellPolicyDecision,
#228	*,
#229	timeout: float,
#230	cwd: str \| os.PathLike[str] \| None,
#231	) -> None:
#232	if self.human is None:
#233	raise CapabilityDenied(f"{pid} requires human approval for shell execute on {resource}")
#234	request_id = self.human.query(
#235	pid=pid,
#236	human=self.config.runtime.default_human,
#237	request={
#238	"type": "external_operation_approval",
#239	"question": f"Allow this process to run shell command {argv[0]!r}?",
#240	"requested_once_capability": {
#241	"subject": pid,
#242	"resource": resource,
#243	"rights": [CapabilityRight.EXECUTE.value],
#244	},
#245	"context": {
#246	"adapter": "shell",
#247	"primitive": "runtime.shell.run",
#248	"operation": "run",
#249	"pid": pid,
#250	"workspace_root": str(getattr(self.provider, "cwd", "")),
#251	"working_directory": os.fspath(cwd) if cwd is not None else ".",
#252	"argv": list(argv),
#253	"command": argv[0],
#254	"resource": resource,
#255	"right": CapabilityRight.EXECUTE.value,
#256	"grant_scope": "one_time",
#257	"timeout_s": timeout,
#258	"policy_level": decision.policy_level,
#259	"policy_reason": decision.reason,
#260	"matched_rule": list(decision.matched_rule) if decision.matched_rule else None,
#261	"high_risk": decision.high_risk,
#262	},
#263	},
#264	blocking=True,
#265	)
#266	raise HumanApprovalRequired(
#267	request_id=request_id,
#268	message=f"{pid} is waiting for per-use human approval to run {resource}",
#269	)
#270
#271	def _emit_run_event(
#272	self,
#273	pid: str,
#274	resource: str,
#275	argv: list[str],
#276	proc: CommandResult,
#277	decision: ShellPolicyDecision,
#278	*,
#279	cwd: str \| os.PathLike[str] \| None,
#280	) -> None:
#281	if self.events is None:
#282	return
#283	self.events.emit(
#284	EventType.EXTERNAL_WRITE,
#285	source=pid,
#286	target=resource,
#287	payload={
#288	"adapter": "shell",
#289	"operation": "run",
#290	"argv": argv,
#291	"returncode": proc.returncode,
#292	"policy_level": decision.policy_level,
#293	"high_risk": decision.high_risk,
#294	"cwd": os.fspath(cwd) if cwd is not None else None,
#295	},
#296	)
#297
#298	def _matching_shell_policy_caps(self, pid: str, resource: str) -> list[Capability]:
#299	caps = [
#300	cap
#301	for cap in self._matching_shell_caps(pid, resource)
#302	if self.config.shell.policy_capability_key in cap.constraints
#303	]
#304	caps.sort(key=lambda cap: (len(cap.resource), cap.issued_at), reverse=True)
#305	return caps
#306
#307	def _legacy_permission_policy(self, pid: str, resource: str) -> tuple[str, bool]:
#308	caps = [
#309	cap
#310	for cap in self._matching_shell_caps(pid, resource)
#311	if self.config.shell.policy_capability_key not in cap.constraints
#312	]
#313	if not caps:
#314	return CapabilityManager.MISSING, False
#315	caps.sort(key=lambda cap: (len(cap.resource), cap.issued_at), reverse=True)
#316	cap = caps[0]
#317	policy = str(cap.constraints.get(CapabilityManager.POLICY_KEY) or CapabilityManager.ALWAYS_ALLOW)
#318	return policy, policy == CapabilityManager.ALLOW_ONCE
#319
#320	def _matching_shell_caps(self, pid: str, resource: str) -> list[Capability]:
#321	now = utc_now()
#322	result: list[Capability] = []
#323	for cap in self.capabilities.capabilities_for(pid):
#324	if cap.revoked:
#325	continue
#326	if cap.expires_at is not None and cap.expires_at <= now:
#327	continue
#328	if CapabilityRight.EXECUTE.value not in cap.rights and "*" not in cap.rights:
#329	continue
#330	if not self._resource_matches(cap.resource, resource):
#331	continue
#332	result.append(cap)
#333	return result
#334
#335	def _selected_policy_level(self, caps: list[Capability]) -> str:
#336	return self._normalize_policy_level(caps[0].constraints[self.config.shell.policy_capability_key])
#337
#338	def _normalize_policy_level(self, value: Any) -> str:
#339	normalized = str(value).strip().lower()
#340	allowed = {
#341	self.config.shell.always_deny_level,
#342	self.config.shell.allowlist_auto_else_ask_level,
#343	self.config.shell.blocklist_ask_else_auto_level,
#344	self.config.shell.always_allow_level,
#345	}
#346	if normalized not in allowed:
#347	raise ValidationError(f"unknown shell policy level: {value!r}")
#348	return normalized
#349
#350	def _first_matching_blacklist_rule(self, argv: list[str]) -> ShellCommandRule \| None:
#351	direct = self._first_matching_rule(argv, self.config.shell.blacklist, allow_bare_only=False)
#352	if direct is not None:
#353	return direct
#354	# Some executables such as env/nohup/sudo can dispatch another program.
#355	# In blacklist mode, scan later tokens for exact executable identities
#356	# so `env bash -c ...` still requires human approval.
#357	executable_tokens = {self._normalize_executable(rule.argv[0]) for rule in self.config.shell.blacklist}
#358	for token in argv[1:]:
#359	if self._normalize_executable(token) in executable_tokens:
#360	return ShellCommandRule((token,), match="exact", description="nested blacklist executable")
#361	return None
#362
#363	def _first_matching_rule(
#364	self,
#365	argv: list[str],
#366	rules: tuple[ShellCommandRule, ...],
#367	*,
#368	allow_bare_only: bool,
#369	) -> ShellCommandRule \| None:
#370	return next((rule for rule in rules if self._rule_matches(argv, rule, allow_bare_only=allow_bare_only)), None)
#371
#372	def _rule_matches(self, argv: list[str], rule: ShellCommandRule, *, allow_bare_only: bool) -> bool:
#373	if not rule.argv:
#374	return False
#375	if allow_bare_only and self._argv0_has_path(argv[0]) and not self._argv0_has_path(rule.argv[0]):
#376	return False
#377	if rule.match == "exact" and len(argv) != len(rule.argv):
#378	return False
#379	if len(argv) < len(rule.argv):
#380	return False
#381	for index, expected in enumerate(rule.argv):
#382	actual = argv[index]
#383	if index == 0:
#384	if self._normalize_executable(actual) != self._normalize_executable(expected):
#385	return False
#386	continue
#387	if actual != expected:
#388	return False
#389	return True
#390
#391	def _validate_argv(self, argv: list[str]) -> list[str]:
#392	if not isinstance(argv, list) or not argv:
#393	raise ValidationError("shell argv must be a non-empty list")
#394	checked: list[str] = []
#395	for index, value in enumerate(argv):
#396	if not isinstance(value, str):
#397	raise ValidationError(f"shell argv[{index}] must be a string")
#398	if "\x00" in value:
#399	raise ValidationError(f"shell argv[{index}] cannot contain NUL bytes")
#400	if index == 0 and not value.strip():
#401	raise ValidationError("shell argv[0] must be non-empty")
#402	checked.append(value)
#403	return checked
#404
#405	def _validate_timeout(self, timeout: float) -> float:
#406	try:
#407	selected = float(timeout)
#408	except (TypeError, ValueError) as exc:
#409	raise ValidationError("shell timeout must be a number") from exc
#410	if selected <= 0:
#411	raise ValidationError("shell timeout must be > 0")
#412	if selected > self.config.shell.timeout_hard_limit_s:
#413	raise ValidationError(f"shell timeout exceeds hard limit {self.config.shell.timeout_hard_limit_s}s")
#414	return selected
#415
#416	def _bounded_result(self, proc: CommandResult) -> CommandResult:
#417	stdout, stdout_truncated = self._truncate_output(proc.stdout, self.config.shell.max_stdout_chars)
#418	stderr, stderr_truncated = self._truncate_output(proc.stderr, self.config.shell.max_stderr_chars)
#419	return CommandResult(
#420	argv=proc.argv,
#421	returncode=proc.returncode,
#422	stdout=stdout,
#423	stderr=stderr,
#424	stdout_truncated=proc.stdout_truncated or stdout_truncated,
#425	stderr_truncated=proc.stderr_truncated or stderr_truncated,
#426	)
#427
#428	def _truncate_output(self, value: str, limit: int) -> tuple[str, bool]:
#429	if len(value) <= limit:
#430	return value, False
#431	return value[:limit], True
#432
#433	def _command_identity(self, argv0: str) -> str:
#434	return self._normalize_executable(argv0)
#435
#436	def _normalize_executable(self, value: str) -> str:
#437	raw = value.strip().replace("\\", "/")
#438	name = PurePath(raw).name or raw
#439	lowered = name.casefold()
#440	for suffix in _WINDOWS_EXECUTABLE_SUFFIXES:
#441	if lowered.endswith(suffix):
#442	return lowered[: -len(suffix)]
#443	return lowered
#444
#445	def _argv0_has_path(self, value: str) -> bool:
#446	return "/" in value or "\\" in value or PurePath(value).is_absolute()
#447
#448	def _resource_matches(self, granted: str, requested: str) -> bool:
#449	if granted == "*" or granted == requested:
#450	return True
#451	if granted.endswith(":*"):
#452	return requested.startswith(granted[:-1])
#453	return False
#454

z6MkqRzA/yingqi-z20-Agent-libOS