| """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.") |
|
|
