yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	from dataclasses import replace
#4	from typing import Iterable
#5
#6	from agent_libos.config import DEFAULT_CONFIG, AgentLibOSConfig
#7	from agent_libos.models.exceptions import CapabilityDenied, NotFound
#8	from agent_libos.utils.ids import new_id, utc_now
#9	from agent_libos.models import Capability, CapabilityRight, EventType, ObjectHandle
#10	from agent_libos.runtime.audit_manager import AuditManager
#11	from agent_libos.runtime.event_bus import EventBus
#12	from agent_libos.storage import SQLiteStore
#13
#14
#15	class CapabilityManager:
#16	"""Capability directory and permission-policy helper."""
#17
#18	POLICY_KEY = "permission_policy"
#19	ALWAYS_ALLOW = "always_allow"
#20	ALWAYS_DENY = "always_deny"
#21	ASK_EACH_TIME = "ask_each_time"
#22	ALLOW_ONCE = "allow_once"
#23	MISSING = "missing"
#24	POLICY_VALUES = {ALWAYS_ALLOW, ALWAYS_DENY, ASK_EACH_TIME, ALLOW_ONCE}
#25
#26	def __init__(self, store: SQLiteStore, audit: AuditManager, events: EventBus, config: AgentLibOSConfig \| None = None):
#27	self.config = config or DEFAULT_CONFIG
#28	self.store = store
#29	self.audit = audit
#30	self.events = events
#31
#32	def grant(
#33	self,
#34	subject: str,
#35	resource: str,
#36	rights: Iterable[str \| CapabilityRight],
#37	issued_by: str = "system",
#38	constraints: dict \| None = None,
#39	expires_at: str \| None = None,
#40	delegable: bool = False,
#41	revocable: bool = True,
#42	) -> Capability:
#43	cap = Capability(
#44	cap_id=new_id("cap"),
#45	subject=subject,
#46	resource=resource,
#47	rights={str(right) for right in rights},
#48	constraints=constraints or {},
#49	issued_by=issued_by,
#50	issued_at=utc_now(),
#51	expires_at=expires_at,
#52	delegable=delegable,
#53	revocable=revocable,
#54	revoked=False,
#55	)
#56	self.store.insert_capability(cap)
#57	self._attach_to_process(subject, cap.cap_id)
#58	self.events.emit(
#59	EventType.CAPABILITY_GRANTED,
#60	source=issued_by,
#61	target=subject,
#62	payload={"capability_id": cap.cap_id, "resource": resource, "rights": sorted(cap.rights)},
#63	)
#64	self.audit.record(
#65	actor=issued_by,
#66	action="capability.grant",
#67	target=f"{subject}:{resource}",
#68	capability_refs=[cap.cap_id],
#69	decision={"granted": True, "rights": sorted(cap.rights)},
#70	)
#71	return cap
#72
#73	def revoke(self, cap_id: str, revoked_by: str = "system", reason: str \| None = None) -> Capability:
#74	cap = self.store.get_capability(cap_id)
#75	if cap is None:
#76	raise NotFound(f"capability not found: {cap_id}")
#77	if not cap.revocable:
#78	raise CapabilityDenied(f"capability is not revocable: {cap_id}")
#79	revoked = replace(cap, revoked=True)
#80	self.store.update_capability(revoked)
#81	self.events.emit(
#82	EventType.CAPABILITY_REVOKED,
#83	source=revoked_by,
#84	target=cap.subject,
#85	payload={"capability_id": cap_id, "reason": reason},
#86	)
#87	self.audit.record(
#88	actor=revoked_by,
#89	action="capability.revoke",
#90	target=cap.resource,
#91	capability_refs=[cap_id],
#92	decision={"revoked": True, "reason": reason},
#93	)
#94	return revoked
#95
#96	def check(self, subject: str, resource: str, right: str \| CapabilityRight) -> bool:
#97	# ASK_EACH_TIME is intentionally not treated as allowed here; only the
#98	# primitive with enough operation context may turn it into a human prompt.
#99	return self.permission_policy(subject, resource, right) in {self.ALWAYS_ALLOW, self.ALLOW_ONCE}
#100
#101	def require(self, subject: str, resource: str, right: str \| CapabilityRight) -> None:
#102	requested_right = str(right)
#103	policy = self.permission_policy(subject, resource, right)
#104	if policy in {self.ALWAYS_ALLOW, self.ALLOW_ONCE}:
#105	return
#106	if policy == self.ALWAYS_DENY:
#107	raise CapabilityDenied(f"{subject} denied {requested_right} on {resource}")
#108	if policy == self.ASK_EACH_TIME:
#109	raise CapabilityDenied(f"{subject} requires human approval for {requested_right} on {resource}")
#110	raise CapabilityDenied(f"{subject} lacks {requested_right} on {resource}")
#111
#112	def permission_policy(self, subject: str, resource: str, right: str \| CapabilityRight) -> str:
#113	matches = self._matching_capabilities(subject, resource, right)
#114	if not matches:
#115	return self.MISSING
#116	cap = matches[0]
#117	return str(cap.constraints.get(self.POLICY_KEY) or self.ALWAYS_ALLOW)
#118
#119	def set_permission_policy(
#120	self,
#121	subject: str,
#122	resource: str,
#123	rights: Iterable[str \| CapabilityRight],
#124	policy: str,
#125	issued_by: str \| None = None,
#126	constraints: dict \| None = None,
#127	) -> Capability:
#128	if policy not in self.POLICY_VALUES:
#129	raise ValueError(f"unknown permission policy: {policy}")
#130	merged_constraints = dict(constraints or {})
#131	merged_constraints[self.POLICY_KEY] = policy
#132	cap = self.grant(
#133	subject=subject,
#134	resource=resource,
#135	rights=rights,
#136	issued_by=issued_by or self.config.runtime.default_human_actor,
#137	constraints=merged_constraints,
#138	)
#139	actor = issued_by or self.config.runtime.default_human_actor
#140	self.audit.record(
#141	actor=actor,
#142	action="capability.permission_policy",
#143	target=f"{subject}:{resource}",
#144	capability_refs=[cap.cap_id],
#145	decision={"policy": policy, "rights": sorted(cap.rights)},
#146	)
#147	return cap
#148
#149	def grant_once(
#150	self,
#151	subject: str,
#152	resource: str,
#153	rights: Iterable[str \| CapabilityRight],
#154	issued_by: str \| None = None,
#155	constraints: dict \| None = None,
#156	) -> Capability:
#157	return self.set_permission_policy(
#158	subject=subject,
#159	resource=resource,
#160	rights=rights,
#161	policy=self.ALLOW_ONCE,
#162	issued_by=issued_by or self.config.runtime.default_human_actor,
#163	constraints=constraints,
#164	)
#165
#166	def inherit(
#167	self,
#168	parent: str,
#169	child: str,
#170	resource: str,
#171	rights: Iterable[str \| CapabilityRight],
#172	issued_by: str,
#173	constraints: dict \| None = None,
#174	) -> Capability:
#175	normalized = {str(right) for right in rights}
#176	if not normalized:
#177	raise ValueError("inherited capability must include at least one right")
#178	for right in normalized:
#179	policy = self.permission_policy(parent, resource, right)
#180	if policy != self.ALWAYS_ALLOW:
#181	raise CapabilityDenied(
#182	f"{parent} cannot inherit {right} on {resource} to {child}; parent policy is {policy}"
#183	)
#184	inherited_constraints = dict(constraints or {})
#185	inherited_constraints[self.POLICY_KEY] = self.ALWAYS_ALLOW
#186	inherited_constraints["inherited_from"] = parent
#187	cap = self.grant(
#188	subject=child,
#189	resource=resource,
#190	rights=normalized,
#191	issued_by=issued_by,
#192	constraints=inherited_constraints,
#193	)
#194	self.audit.record(
#195	actor=issued_by,
#196	action="capability.inherit",
#197	target=f"{parent}->{child}:{resource}",
#198	capability_refs=[cap.cap_id],
#199	decision={"rights": sorted(normalized)},
#200	)
#201	return cap
#202
#203	def consume_allow_once(self, subject: str, resource: str, right: str \| CapabilityRight, used_by: str) -> None:
#204	# One-shot grants are consumed by the primitive after the operation
#205	# succeeds, so a failed attempt does not burn the user's approval.
#206	for cap in self._matching_capabilities(subject, resource, right):
#207	if cap.constraints.get(self.POLICY_KEY) == self.ALLOW_ONCE:
#208	self.revoke(cap.cap_id, revoked_by=used_by, reason="one-time permission consumed")
#209	return
#210
#211	def assert_handle(self, subject: str, handle: ObjectHandle, right: str \| CapabilityRight) -> None:
#212	requested = str(right)
#213	if requested not in handle.rights and "*" not in handle.rights:
#214	raise CapabilityDenied(f"object handle lacks {requested}: {handle.oid}")
#215	cap = self.store.get_capability(handle.capability_id)
#216	if cap is None or cap.revoked:
#217	raise CapabilityDenied(f"invalid object capability: {handle.capability_id}")
#218	if cap.subject != subject:
#219	raise CapabilityDenied(f"capability subject mismatch: {cap.subject} != {subject}")
#220	if not self._resource_matches(cap.resource, f"object:{handle.oid}"):
#221	raise CapabilityDenied(f"capability does not target object: {handle.oid}")
#222	if requested not in cap.rights and "*" not in cap.rights:
#223	raise CapabilityDenied(f"capability lacks {requested}: {handle.oid}")
#224
#225	def handle_for_object(
#226	self,
#227	subject: str,
#228	oid: str,
#229	rights: Iterable[str \| CapabilityRight],
#230	issued_by: str = "system",
#231	expires_at: str \| None = None,
#232	) -> ObjectHandle:
#233	normalized = {str(right) for right in rights}
#234	cap = self.grant(
#235	subject=subject,
#236	resource=f"object:{oid}",
#237	rights=normalized,
#238	issued_by=issued_by,
#239	expires_at=expires_at,
#240	delegable=False,
#241	)
#242	return ObjectHandle(oid=oid, rights=normalized, capability_id=cap.cap_id, expires_at=expires_at)
#243
#244	def capabilities_for(self, subject: str) -> list[Capability]:
#245	return self.store.list_capabilities(subject=subject)
#246
#247	def spec(
#248	self,
#249	resource: str,
#250	rights: Iterable[str \| CapabilityRight],
#251	**kwargs,
#252	) -> dict:
#253	return {"resource": resource, "rights": [str(right) for right in rights], **kwargs}
#254
#255	def tool_execute(self, tool: str, rights: Iterable[str \| CapabilityRight] \| None = None, **kwargs) -> dict:
#256	resource = tool if tool.startswith("tool:") else f"tool:{tool}"
#257	return self.spec(resource, rights or [CapabilityRight.EXECUTE], **kwargs)
#258
#259	def project_read(self, name: str, **kwargs) -> dict:
#260	return self.spec(f"project:{name}", [CapabilityRight.READ], **kwargs)
#261
#262	def object_access(self, oid: str, rights: Iterable[str \| CapabilityRight], **kwargs) -> dict:
#263	return self.spec(f"object:{oid}", rights, **kwargs)
#264
#265	def _attach_to_process(self, subject: str, cap_id: str) -> None:
#266	process = self.store.get_process(subject)
#267	if process is None:
#268	return
#269	if cap_id not in process.capabilities:
#270	process.capabilities.append(cap_id)
#271	process.updated_at = utc_now()
#272	self.store.update_process(process)
#273
#274	def _resource_matches(self, granted: str, requested: str) -> bool:
#275	if granted == "*" or granted == requested:
#276	return True
#277	if granted.endswith(":*"):
#278	return requested.startswith(granted[:-1])
#279	if granted.endswith("/*"):
#280	return requested.startswith(granted[:-1])
#281	return False
#282
#283	def _matching_capabilities(self, subject: str, resource: str, right: str \| CapabilityRight) -> list[Capability]:
#284	requested_right = str(right)
#285	now = utc_now()
#286	matches: list[Capability] = []
#287	for cap in self.store.list_capabilities(subject=subject):
#288	if cap.revoked:
#289	continue
#290	if cap.expires_at is not None and cap.expires_at <= now:
#291	continue
#292	if not self._resource_matches(cap.resource, resource):
#293	continue
#294	if "*" not in cap.rights and requested_right not in cap.rights:
#295	continue
#296	matches.append(cap)
#297	# Prefer the most specific resource grant when overlapping wildcard and
#298	# path-level capabilities both exist.
#299	matches.sort(key=lambda cap: (len(cap.resource), cap.issued_at), reverse=True)
#300	return matches
#301

z6MkqRzA/yingqi-z20-Agent-libOS