"""Per-agent policy files for Egress Guard. A policy file says, for one agent, which page types it may open, which domains are always allowed or denied, what happens to destinations nobody has classified, and which narrow exceptions exist. It sits on top of a Guard: the Guard decides what a URL is and whether the baseline policy denies it; the agent policy then tightens or, through a named exception, loosens that decision for this agent only. from agent_egress_guard import Guard, AgentPolicy policy = AgentPolicy.load("vendor-research.json") v = policy.check(Guard(api_key="..."), "https://vendor.example/start-trial", "GET") v.decision # "deny", "allow" or "approval_required" Policy file (JSON; YAML also works when PyYAML is installed): { "agent": "vendor-research", "owner": "head-of-procurement", "tier": 3, "review_by": "2026-12-31", "web": { "allow_page_types": ["pricing", "documentation", "about", "legal", "security", "status"], "deny_page_types": ["careers"], "allow_domains": ["approved-supplier.example"], "deny_domains": ["competitor.example"], "unclassified": "deny", "on_deny": "ask_owner", "exceptions": [ {"page_type": "checkout", "domain": "approved-supplier.example", "expires": "2027-03-31", "requires_approval": true} ] } } """ import datetime as _dt import json import os from dataclasses import replace from urllib.parse import urlsplit from .engine import DENY_TYPES, Verdict, _canon_host # the 28 page types of the AI Agent Allowlist database READ_TYPES = {"contact", "about", "legal", "product", "blog", "help_center", "pricing", "events", "sitemap", "careers", "case_studies", "leadership", "partners", "press", "documentation", "community", "security", "status", "integrations"} ACTION_TYPES = {"signup", "password_reset", "cart", "checkout", "upload", "post_create", "comment", "subscribe"} PAGE_TYPES = READ_TYPES | ACTION_TYPES | {"login"} UNCLASSIFIED = ("allow_reads", "deny") ON_DENY = ("block", "ask_owner") class PolicyError(ValueError): """The policy file cannot be used.""" def _today(): return _dt.date.today() def _date(s, field_name): try: return _dt.date.fromisoformat(str(s)) except ValueError: raise PolicyError("%s must be a date like 2026-12-31, got %r" % (field_name, s)) def _domain_match(host, domain): host = host[4:] if host.startswith("www.") else host domain = domain.lower().strip().lstrip(".") domain = domain[4:] if domain.startswith("www.") else domain return host == domain or host.endswith("." + domain) def _read_file(path): with open(path, encoding="utf-8") as fh: text = fh.read() if path.lower().endswith((".yaml", ".yml")): try: import yaml # optional dependency except ImportError: raise PolicyError("YAML policy files need PyYAML (pip install pyyaml), or save the policy as JSON") return yaml.safe_load(text) return json.loads(text) class AgentPolicy: def __init__(self, data): if not isinstance(data, dict): raise PolicyError("a policy must be a JSON object") self.data = data self.agent = str(data.get("agent") or "").strip() self.owner = str(data.get("owner") or "").strip() self.tier = data.get("tier") self.review_by = _date(data["review_by"], "review_by") if data.get("review_by") else None web = data.get("web") or {} if not isinstance(web, dict): raise PolicyError("'web' must be an object") self.allow_types = set(web.get("allow_page_types") or []) self.deny_types = set(web.get("deny_page_types") or []) self.allow_domains = [d.lower() for d in web.get("allow_domains") or []] self.deny_domains = [d.lower() for d in web.get("deny_domains") or []] self.unclassified = web.get("unclassified", "allow_reads") self.on_deny = web.get("on_deny", "block") self.exceptions = [] for i, e in enumerate(web.get("exceptions") or []): if not isinstance(e, dict) or not e.get("page_type") or not e.get("domain"): raise PolicyError("exception %d needs page_type and domain" % (i + 1)) self.exceptions.append({ "page_type": e["page_type"], "domain": str(e["domain"]).lower(), "expires": _date(e["expires"], "exceptions[%d].expires" % i) if e.get("expires") else None, "requires_approval": bool(e.get("requires_approval", False)), }) problems = [p for p in self.validate() if p.startswith("error:")] if problems: raise PolicyError("; ".join(p[7:] for p in problems)) @classmethod def load(cls, path): if not os.path.exists(path): raise PolicyError("policy file not found: %s" % path) try: return cls(_read_file(path)) except json.JSONDecodeError as e: raise PolicyError("not valid JSON: %s" % e) def validate(self, today=None): """List of 'error: ...' and 'warning: ...' strings. Errors make the policy unusable.""" today = today or _today() out = [] if self.unclassified not in UNCLASSIFIED: out.append("error: web.unclassified must be one of %s" % ", ".join(UNCLASSIFIED)) if self.on_deny not in ON_DENY: out.append("error: web.on_deny must be one of %s" % ", ".join(ON_DENY)) for t in sorted((self.allow_types | self.deny_types) - PAGE_TYPES): out.append("error: unknown page type %r" % t) for t in sorted(self.allow_types & self.deny_types): out.append("error: page type %r is both allowed and denied" % t) for t in sorted(self.allow_types & (ACTION_TYPES | {"login"})): out.append("error: %r is an action page type; grant it through an exception for one domain instead" % t) for e in self.exceptions: if e["page_type"] not in PAGE_TYPES: out.append("error: exception for unknown page type %r" % e["page_type"]) if e["expires"] is None: out.append("warning: exception %s on %s has no expiry date" % (e["page_type"], e["domain"])) elif e["expires"] < today: out.append("warning: exception %s on %s expired on %s and is ignored" % (e["page_type"], e["domain"], e["expires"])) if not self.agent: out.append("warning: no agent name") if not self.owner: out.append("warning: no owner; every agent should have one named owner") if self.review_by is None: out.append("warning: no review_by date") elif self.review_by < today: out.append("warning: review overdue since %s" % self.review_by) if self.tier in (3, 4) and self.unclassified != "deny": out.append("warning: tier %s agents should deny unclassified destinations" % self.tier) return out def _exception_for(self, host, page_type, today): for e in self.exceptions: if e["page_type"] == page_type and _domain_match(host, e["domain"]): if e["expires"] is None or e["expires"] >= today: return e return None def apply(self, v, today=None): """Apply this agent's policy to a Guard verdict and return the final verdict.""" today = today or _today() host = _canon_host(urlsplit(v.url if "://" in v.url else "https://" + v.url).hostname or "") tag = {"agent": self.agent} if self.agent else {} base = {"base_decision": v.decision, "base_layer": v.layer, "base_rule": v.rule} def out(decision, rule, note, page_type=None): return Verdict(decision, "agent_policy", rule, v.url, v.method, page_type or v.page_type, note, dict(v.extra, **base, **tag)) if any(_domain_match(host, d) for d in self.deny_domains): return out("deny", "deny_domain", "domain denied for this agent") ptype = v.page_type or (v.rule if v.layer == "page_type_db" else "") if v.decision == "deny": e = self._exception_for(host, ptype, today) if ptype else None if e: if e["requires_approval"]: return out("approval_required", "exception", "exception needs owner approval per action") return out("allow", "exception", "allowed by a named exception until %s" % (e["expires"] or "no end date")) if self.on_deny == "ask_owner" and v.layer != "high_value_hosts": return out("approval_required", v.rule, "denied by policy; routed to the agent owner") return v if ptype and ptype in self.deny_types: return out("deny", "deny_page_type", "page type %s denied for this agent" % ptype) if v.layer == "page_type_db" and self.allow_types and ptype not in self.allow_types: return out("deny", "page_type_not_allowed", "page type %s is not on this agent's allow list" % ptype) if v.layer == "default" and v.decision == "allow" and self.unclassified == "deny": if not any(_domain_match(host, d) for d in self.allow_domains): return out("deny", "unclassified", "destination not classified; this agent denies unclassified pages") return replace(v, extra=dict(v.extra, **tag)) def check(self, guard, url, method="GET", today=None): return self.apply(guard.check(url, method), today) def policy_template(agent="my-agent", owner="", tier=3): """A starting policy: read pages allowed by purpose, every action page denied, unclassified denied.""" return { "agent": agent, "owner": owner, "tier": tier, "review_by": (_today() + _dt.timedelta(days=90)).isoformat(), "web": { "allow_page_types": sorted(READ_TYPES - {"careers"}), "deny_page_types": [], "allow_domains": [], "deny_domains": [], "unclassified": "deny" if tier in (3, 4) else "allow_reads", "on_deny": "ask_owner" if tier == 3 else "block", "exceptions": [], }, } __all__ = ["AgentPolicy", "PolicyError", "policy_template", "PAGE_TYPES", "READ_TYPES", "ACTION_TYPES", "DENY_TYPES"]