"""YAML capability manifest loading and validation."""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, cast
from urllib.parse import urlsplit
import jsonschema
import yaml
_SCHEMA_NAME = "capability_manifest.schema.json"
_DEFAULT_PORTS = {"http": 80, "https": 443}
def _endpoint_matches(endpoint: str, allowed: str) -> bool:
"""URL-aware allow-list match: exact scheme/host/port, prefix on path.
Plain string prefixes are unsafe here — ``https://api.example.com``
would also match ``https://api.example.com.attacker.net`` and
``https://api.example.com@attacker.net``.
"""
try:
target, rule = urlsplit(endpoint), urlsplit(allowed)
target_port, rule_port = target.port, rule.port
except ValueError:
return False
if not rule.scheme or not rule.hostname:
return False # malformed allow-list entries never match
if target.scheme.lower() != rule.scheme.lower():
return False
if (target.hostname or "").lower() != rule.hostname.lower():
return False
default_port = _DEFAULT_PORTS.get(rule.scheme.lower())
if (target_port or default_port) != (rule_port or default_port):
return False
if rule.path in ("", "/"):
return True
if rule.path.endswith("/"):
return target.path.startswith(rule.path)
return target.path == rule.path or target.path.startswith(rule.path + "/")
def _schema_path() -> Path:
"""Resolve JSON Schema from the installed package (pip-safe)."""
package_root = Path(__file__).resolve().parents[1]
bundled = package_root / "schemas" / _SCHEMA_NAME
if bundled.is_file():
return bundled
# Legacy repo layout: schemas/ at repository root (editable checkouts)
repo_root = package_root.parent
legacy = repo_root / "schemas" / _SCHEMA_NAME
if legacy.is_file():
return legacy
msg = (
f"Capability manifest schema not found. Tried: {bundled}, {legacy}. "
"Reinstall agentguard or ensure agentguard/schemas/ is present."
)
raise FileNotFoundError(msg)
def _load_schema() -> dict[str, Any]:
with _schema_path().open(encoding="utf-8") as handle:
return cast(dict[str, Any], json.load(handle))
[docs]
@dataclass
class CapabilityManifest:
"""Declarative runtime capability scope for an agent."""
agent_id: str
permitted_tools: list[str] = field(default_factory=list)
forbidden_tools: list[str] = field(default_factory=list)
allowed_data_sources: list[str] = field(default_factory=list)
permitted_endpoints: list[str] = field(default_factory=list)
max_output_tokens: int = 4096
external_contact: bool = False
can_spawn_agents: bool = False
max_delegation_depth: int = 0
[docs]
@classmethod
def from_yaml(cls, path: str) -> CapabilityManifest:
"""Load and validate a manifest from a YAML file.
Args:
path: Path to the YAML manifest file.
Returns:
Validated CapabilityManifest instance.
Raises:
jsonschema.ValidationError: If validation fails.
ValueError: If YAML content is invalid.
"""
yaml_path = Path(path)
with yaml_path.open(encoding="utf-8") as handle:
data = yaml.safe_load(handle)
if not isinstance(data, dict):
msg = f"Invalid manifest YAML in {yaml_path}"
raise ValueError(msg)
jsonschema.validate(instance=data, schema=_load_schema())
return cls.from_dict(data)
[docs]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> CapabilityManifest:
"""Build a manifest from an already-validated dictionary."""
return cls(
agent_id=str(data["agent_id"]),
permitted_tools=list(data.get("permitted_tools", [])),
forbidden_tools=list(data.get("forbidden_tools", [])),
allowed_data_sources=list(data.get("allowed_data_sources", [])),
permitted_endpoints=list(data.get("permitted_endpoints", [])),
max_output_tokens=int(data.get("max_output_tokens", 4096)),
external_contact=bool(data.get("external_contact", False)),
can_spawn_agents=bool(data.get("can_spawn_agents", False)),
max_delegation_depth=int(data.get("max_delegation_depth", 0)),
)
[docs]
def is_endpoint_permitted(self, endpoint: str) -> bool:
"""Return True if external contact is allowed and the endpoint matches.
Matching is URL-aware: scheme, host, and port must match an
allow-list entry exactly; the path is matched as a segment-safe
prefix (``/v1`` permits ``/v1/reports`` but not ``/v1evil``).
"""
if not self.external_contact:
return False
if not self.permitted_endpoints:
return False
return any(_endpoint_matches(endpoint, allowed) for allowed in self.permitted_endpoints)
[docs]
def is_data_source_allowed(self, data_source: str) -> bool:
"""Return True if the data source is explicitly listed.
An empty ``allowed_data_sources`` list is deny-by-default so that
attenuation intersections cannot accidentally expand access.
"""
return data_source in self.allowed_data_sources
[docs]
def attenuate(self, other: CapabilityManifest) -> CapabilityManifest:
"""Return monotonic attenuation: intersection of both manifests.
Args:
other: Sub-agent manifest to attenuate against this manifest.
Returns:
New manifest with reduced (never expanded) permissions.
"""
permitted = sorted(set(self.permitted_tools) & set(other.permitted_tools))
forbidden = sorted(set(self.forbidden_tools) | set(other.forbidden_tools))
permitted = [tool for tool in permitted if tool not in forbidden]
return CapabilityManifest(
agent_id=other.agent_id,
permitted_tools=permitted,
forbidden_tools=forbidden,
allowed_data_sources=sorted(
set(self.allowed_data_sources) & set(other.allowed_data_sources),
),
permitted_endpoints=sorted(
set(self.permitted_endpoints) & set(other.permitted_endpoints),
),
max_output_tokens=min(self.max_output_tokens, other.max_output_tokens),
external_contact=self.external_contact and other.external_contact,
can_spawn_agents=self.can_spawn_agents and other.can_spawn_agents,
max_delegation_depth=min(self.max_delegation_depth, other.max_delegation_depth),
)