Human-in-the-loop TOTP approval for dangerous AI-agent actions. Zero dependencies — Python standard library only.
You gave your agent shell access, file writes, e-mail, maybe money. At some point it will want to do something you'd rather personally sign off on — from your phone, while away from the machine. The obvious answer ("just reply yes in the chat") is the wrong one:
- chat sessions get hijacked and messages get spoofed,
- old approvals can be replayed from history,
- and an LLM can be prompt-injected into believing approval arrived.
A valid TOTP code (the 6 digits from Google Authenticator / Authy / 1Password) proves the approver currently holds the enrolled device. It cannot be faked from message history, and each code approves at most one action.
This pattern has been running in production in my personal agent
(23 dangerous tools behind it — process kill, file write, e-mail,
deploys) with Telegram as the relay channel. totp-gate is the
extracted, channel-agnostic core.
pip install totp-gatefrom totp_gate import ApprovalGate, generate_secret, provisioning_uri
# One-time setup: generate a secret, enroll it in an authenticator app.
secret = generate_secret()
print(provisioning_uri(secret, "baran", issuer="my-agent")) # → QR code
gate = ApprovalGate(secret=secret, ttl_seconds=300)
# 1. Agent hits a dangerous action → request approval and pause.
action = gate.request("delete_repo", detail="rm -rf legacy/")
notify_human(f"⚠️ approve `{action.name}` with /o <code> [{action.id}]")
# 2. The human replies with the 6-digit code from their phone.
# Your relay (Telegram bot, Slack handler, CLI...) calls:
if gate.approve(action.id, "492817"):
run_dangerous_thing()
else:
pass # stays blocked — wrong code, expired, or replayedThe gate is channel-agnostic: it never talks to Telegram or Slack
itself. Wire request → your outbound notifier and your inbound
command handler → approve / deny, and any channel works.
async def on_message(update, context):
text = update.message.text
if text.startswith("/o "): # "/o 492817"
code = text.split(maxsplit=1)[1]
for action in gate.pending():
if gate.approve(action.id, code):
await update.message.reply_text(f"✅ {action.name} approved")
return
await update.message.reply_text("❌ rejected")| Property | How |
|---|---|
| Fail-closed | Unknown id, expired request, wrong code → False, action stays blocked |
| Single-use requests | Approve/deny removes the action from the queue |
| Replay-proof | A TOTP time step is never accepted twice (RFC 6238 §5.2) |
| Constant-time compare | hmac.compare_digest on codes |
| Expiry | Requests auto-expire after ttl_seconds (default 5 min) |
| Clock drift | window=1 accepts ±1 period (30 s) |
The TOTP implementation is validated against the official RFC 6238
Appendix B test vectors — see tests/test_totp.py.
ApprovalGate(secret, ttl_seconds=300, window=1, clock=time.time).request(name, detail="") -> PendingAction.pending() -> list[PendingAction].approve(action_id, code) -> bool.deny(action_id) -> bool
generate_secret()/provisioning_uri(secret, account, issuer)totp_at(secret, at, ...)/verify(secret, code, ...)— standalone RFC 6238 primitives if you only need TOTP.
Persistence (queue is in-memory by design — restart = deny-all, which is the fail-closed default you want), rate limiting (do it at your relay), and multi-approver policies (may come later).
MIT