File size: 4,924 Bytes
e99aa7b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """AffixIO agent tools: fail-closed local reference (v1.0.0)."""
from __future__ import annotations
import hashlib, json, time, uuid
ALLOWED_TOOLS = {"identity.check", "permissions.check", "payments.validate",
"email.search", "email.organise", "email.draft", "email.send",
"webhook.trigger", "api.call", "audit.show"}
APPROVAL_REQUIRED = {"email.send", "payments.execute", "webhook.trigger"}
ALLOWED_HOSTS = {"api.example.com", "hooks.example.com"}
BLOCKED = ["sk_live_", "AKIA", "PRIVATE KEY", "aio_live_"]
MAX_AMOUNT = 500.0
INBOX = [
{"id": "m1", "from": "boss@acme.com", "subject": "Q3 invoice approval",
"body": "Approve invoice INV-2041, 320 GBP.", "label": "finance"},
{"id": "m2", "from": "news@newsletter.io", "subject": "Deals",
"body": "Weekly deals.", "label": "newsletter"},
]
AUDIT = []
PENDING = {}
def digest(p) -> str:
return hashlib.sha256(json.dumps(p, sort_keys=True).encode()).hexdigest()
def audit_record(decision, tool, args, reason):
e = {"audit_id": "adt_" + uuid.uuid4().hex[:12],
"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"decision": decision, "tool": tool,
"args_hash": digest(args), "reason": reason}
AUDIT.append(e)
return e
def check_identity(agent_id):
ok = bool(agent_id) and agent_id.startswith("agent://")
return ("allow" if ok else "deny", "identity ok" if ok else "unknown agent")
def check_permission(tool, capabilities):
return ("allow" if tool in capabilities else "deny",
"permitted" if tool in capabilities else "tool_not_permitted:" + tool)
def validate_intent(amount, merchant, allowed_merchants=()):
try:
amt = float(amount)
except (TypeError, ValueError):
return ("deny", "amount_invalid")
if amt > MAX_AMOUNT:
return ("deny", "amount above cap")
if allowed_merchants and merchant not in allowed_merchants:
return ("deny", "merchant not allow-listed")
return ("allow", "intent valid")
def verify(tool, args):
if tool not in ALLOWED_TOOLS:
return ("deny", "tool_not_allowed:" + tool)
text = json.dumps(args)
for pat in BLOCKED:
if pat in text:
return ("deny", "blocked secret pattern")
host = args.get("host", "")
if tool in ("webhook.trigger", "api.call") and host and host not in ALLOWED_HOSTS:
return ("deny", "host not allow-listed:" + host)
if tool in APPROVAL_REQUIRED:
return ("review", tool + " requires human approval")
if "amount" in args:
d, r = validate_intent(args.get("amount"), args.get("merchant", ""),
args.get("allowed_merchants", ()))
if d == "deny":
return (d, r)
if tool == "email.send" and not args.get("to"):
return ("deny", "missing_recipient")
return ("allow", "allowed_by_policy")
def run_tool(tool, args):
d, r = verify(tool, args)
audit_record(d, tool, args, r)
if d == "deny":
return "DENIED - %s: %s" % (tool, r)
if d == "review":
pid = "appr_" + uuid.uuid4().hex[:8]
PENDING[pid] = {"tool": tool, "args": args, "status": "pending"}
return "REVIEW - approval id %s" % pid
if tool == "identity.check":
d2, r2 = check_identity(args.get("agent_id", ""))
return "%s: %s" % (d2.upper(), r2)
if tool == "permissions.check":
d2, r2 = check_permission(args.get("tool", ""), args.get("capabilities", []))
return "%s: %s" % (d2.upper(), r2)
if tool == "payments.validate":
d2, r2 = validate_intent(args.get("amount"), args.get("merchant"),
args.get("allowed_merchants", ()))
return "%s: %s" % (d2.upper(), r2)
if tool == "email.search":
q = args.get("query", "").lower()
hits = [m for m in INBOX if q in (m["subject"] + " " + m["body"]).lower()]
return "\n".join("- [%s] %s" % (m["id"], m["subject"]) for m in hits) or "No match."
if tool == "email.organise":
return "Organised: %d messages" % len(INBOX)
if tool == "email.draft":
return "Draft -> %s | %s" % (args.get("to"), args.get("subject"))
if tool == "webhook.trigger":
return "Webhook queued -> %s (simulated)" % args.get("host")
if tool == "api.call":
return "API call -> %s%s (simulated)" % (args.get("host"), args.get("path", "/"))
if tool == "audit.show":
return "Audit entries: %d" % len(AUDIT)
return "Executed (simulated): " + tool
def approve(pid, ok):
it = PENDING.get(pid)
if not it:
return "Unknown approval id."
if it["status"] != "pending":
return "Already %s." % it["status"]
it["status"] = "approved" if ok else "rejected"
audit_record("allow" if ok else "deny", it["tool"], it["args"], "human_" + it["status"])
return ("APPROVED - executed (simulated)." if ok else "REJECTED - not executed.")
|