Skip to content

Permission gate: honor .claude/settings.json allow/ask/deny, serializ… #17

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Revmagi wants to merge 1 commit into
base: main
Choose a base branch
from
+187 −75
Open

Permission gate: honor .claude/settings.json allow/ask/deny, serializ… #17

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
262 changes: 187 additions & 75 deletions backtalk/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,9 @@
called auto-approve, a different axis than the microphone on purpose).
And with permission_mode "ask" (the default), gated tool calls ASK OUT
LOUD and your spoken yes or no decides them; any other answer is
passed back to the agent as the reason.
passed back to the agent as the reason. Tools matched by an "allow"
rule in .claude/settings.json (reads and lookups) skip the spoken ask
entirely; a "deny" rule blocks silently; everything else still asks.

Flags:
--open-mic start in hands-free listening for this session (the
Expand All @@ -55,6 +57,7 @@
import asyncio
import json
import queue
import os
import re
import socket
import sys
Expand Down Expand Up @@ -82,6 +85,14 @@
PERM_TIMEOUT_S = 75
_PERM = {"fut": None, "asked_at": 0.0, # pending ask + when it was posed
"hinted": False} # escape-hatch hint said yet?
# _PERM has exactly one slot: if two tool calls need a decision at the
# same moment (e.g. two tools requested in the same turn), a second
# gate() would overwrite the first one's "fut" before it's answered,
# orphaning it to time out no matter what the user says. This lock
# serializes concurrent asks so only one question is ever live; a
# second caller waits quietly and speaks its own question only after
# the first is fully resolved.
_PERM_LOCK = asyncio.Lock()
_CONFIRM = {"verb": None, "at": 0.0} # pending "say confirm" + when
_INTERRUPT_ANSWER = "\x00interrupt" # sentinel: turn is being killed
# Live AUTO-APPROVE is OUR flag, not an SDK mode flip: the CLI refuses
Expand Down Expand Up @@ -210,88 +221,183 @@ def _full_detail(tool, tool_input, ctx):
return f"use {name}" + (f", {desc[:70]}" if desc else "")


# ---- SETTINGS.JSON-BACKED AUTO-ALLOW. Claude Code's own permission
# rules (allow/ask/deny in .claude/settings.json) already encode the
# split Jamie wants: reads and lookups auto-clear, writes/drafts/sends
# stay gated. The spoken gate below used to ignore that file and ask
# about literally everything; this checks the same rules first, so a
# tool matched by an "allow" rule skips the spoken ask, one matched by
# "deny" is blocked silently, and anything else (an "ask" rule, or no
# rule at all) still falls through to the spoken yes/no — unmatched
# tools stay safe-by-default, never open-by-default.
_RULES_CACHE = {"mtimes": None, "rules": ([], [], [])}


def _rule_paths():
return [
os.path.expanduser("~/.claude/settings.json"),
os.path.join(CFG["agent_dir"], ".claude", "settings.json"),
os.path.join(CFG["agent_dir"], ".claude", "settings.local.json"),
]


def _path_mtime(p):
try:
return os.path.getmtime(p)
except OSError:
return None


def _load_permission_rules():
"""Merge allow/ask/deny arrays across user, project, and project-
local settings files — same load order and array-merge behavior
Claude Code itself uses. Re-checked by mtime each call so an edit
to settings.json takes effect without a restart; missing or
unreadable files just contribute nothing."""
paths = _rule_paths()
mtimes = tuple(_path_mtime(p) for p in paths)
if mtimes == _RULES_CACHE["mtimes"]:
return _RULES_CACHE["rules"]
allow, ask, deny = [], [], []
for p in paths:
try:
with open(p) as f:
data = json.load(f)
except (OSError, ValueError):
continue
perms = data.get("permissions") or {}
allow.extend(perms.get("allow") or [])
ask.extend(perms.get("ask") or [])
deny.extend(perms.get("deny") or [])
_RULES_CACHE["mtimes"] = mtimes
_RULES_CACHE["rules"] = (allow, ask, deny)
return allow, ask, deny


def _rule_matches(rule, tool, tool_input):
"""Same rule syntax Claude Code's own settings use: a bare tool
name ("Read") matches every call to that tool; "Tool(pattern)"
narrows to Bash commands, where a trailing * is a prefix match and
no trailing * requires an exact command match. Other Tool(...)
forms (Edit(path), WebFetch(domain:...)) aren't matched here yet —
they simply never match, which falls through to the spoken ask,
the safe direction."""
if "(" not in rule:
return rule == tool
base, _, rest = rule.partition("(")
if base != tool or tool != "Bash":
return False
pattern = rest[:-1] if rest.endswith(")") else rest
cmd = str((tool_input or {}).get("command", ""))
if pattern.endswith("*"):
return cmd.startswith(pattern[:-1])
return cmd == pattern


def _classify_permission(tool, tool_input):
allow, ask, deny = _load_permission_rules()
if any(_rule_matches(r, tool, tool_input) for r in deny):
return "deny"
if any(_rule_matches(r, tool, tool_input) for r in allow):
return "allow"
return "ask"


def make_permission_gate(mouth):
from claude_agent_sdk import (PermissionResultAllow,
PermissionResultDeny)

async def gate(tool, tool_input, ctx):
if _AUTOAPPROVE["on"]:
return PermissionResultAllow(behavior="allow")
what = _human_what(tool, tool_input, ctx)
detail = _full_detail(tool, tool_input, ctx)
loop = asyncio.get_running_loop()
signals.static_stop()
log(f"[perm] asking: {what}")
log(f"[perm] detail: {detail}")
if tool == "Bash": # the FULL command always reaches the log
log(f"[perm] full command: {str((tool_input or {}).get('command', ''))[:2000]}")
ask = f"Permission check. I want to {what}. Yes, no, or details?"
if not _PERM["hinted"]:
# the escape hatch announces itself exactly once, at the
# moment it becomes relevant (a field case: a new user
# couldn't find the phrase to turn the checks off)
_PERM["hinted"] = True
ask += (" And any time you're done with these checks, say "
"stop asking for permission.")
mouth.say(ask)
answer = None
try:
deadline = loop.time() + PERM_TIMEOUT_S
while answer is None:
fut = loop.create_future()
_PERM["fut"] = fut
_PERM["asked_at"] = time.monotonic()
while True:
try:
got = await asyncio.wait_for(
asyncio.shield(fut), 1.0)
break
except asyncio.TimeoutError:
if loop.time() >= deadline:
fut.cancel()
mouth.say("No answer, so I didn't do it.")
log("[perm] timed out, denied")
return PermissionResultDeny(
behavior="deny",
message="No spoken answer within the "
"timeout; the action was not "
"approved.",
interrupt=False)
# keep the ring honest while we wait
if not mouth.speaking:
signals.set_state("listening")
if (got != _INTERRUPT_ANSWER
and _norm_speech(got) in _DETAILS):
# read the full literal form, then ask again with a
# fresh clock: asking for details is engagement,
# not silence
log("[perm] details requested")
mouth.say(f"The details: I want to {detail}. "
"Yes or no?")
deadline = loop.time() + PERM_TIMEOUT_S
continue
answer = got
finally:
_PERM["fut"] = None
if answer == _INTERRUPT_ANSWER:
log("[perm] turn interrupted, denied silently")
decision = _classify_permission(tool, tool_input)
if decision == "allow":
log(f"[perm] auto-allowed by settings.json: "
f"{_human_what(tool, tool_input, ctx)}")
return PermissionResultAllow(behavior="allow")
if decision == "deny":
log(f"[perm] auto-denied by settings.json: "
f"{_human_what(tool, tool_input, ctx)}")
return PermissionResultDeny(
behavior="deny",
message="Interrupted by the user; the turn is being "
"cancelled.",
message="Blocked by a deny rule in .claude/settings.json.",
interrupt=False)
async with _PERM_LOCK:
what = _human_what(tool, tool_input, ctx)
detail = _full_detail(tool, tool_input, ctx)
loop = asyncio.get_running_loop()
signals.static_stop()
log(f"[perm] asking: {what}")
log(f"[perm] detail: {detail}")
if tool == "Bash": # the FULL command always reaches the log
log(f"[perm] full command: {str((tool_input or {}).get('command', ''))[:2000]}")
ask = f"Permission check. I want to {what}. Yes, no, or details?"
if not _PERM["hinted"]:
# the escape hatch announces itself exactly once, at the
# moment it becomes relevant (a field case: a new user
# couldn't find the phrase to turn the checks off)
_PERM["hinted"] = True
ask += (" And any time you're done with these checks, say "
"stop asking for permission.")
mouth.say(ask)
answer = None
try:
deadline = loop.time() + PERM_TIMEOUT_S
while answer is None:
fut = loop.create_future()
_PERM["fut"] = fut
_PERM["asked_at"] = time.monotonic()
while True:
try:
got = await asyncio.wait_for(
asyncio.shield(fut), 1.0)
break
except asyncio.TimeoutError:
if loop.time() >= deadline:
fut.cancel()
mouth.say("No answer, so I didn't do it.")
log("[perm] timed out, denied")
return PermissionResultDeny(
behavior="deny",
message="No spoken answer within the "
"timeout; the action was not "
"approved.",
interrupt=False)
# keep the ring honest while we wait
if not mouth.speaking:
signals.set_state("listening")
if (got != _INTERRUPT_ANSWER
and _norm_speech(got) in _DETAILS):
# read the full literal form, then ask again with a
# fresh clock: asking for details is engagement,
# not silence
log("[perm] details requested")
mouth.say(f"The details: I want to {detail}. "
"Yes or no?")
deadline = loop.time() + PERM_TIMEOUT_S
continue
answer = got
finally:
_PERM["fut"] = None
if answer == _INTERRUPT_ANSWER:
log("[perm] turn interrupted, denied silently")
return PermissionResultDeny(
behavior="deny",
message="Interrupted by the user; the turn is being "
"cancelled.",
interrupt=False)
approved = _norm_speech(answer) in _YES
# the model keeps working either way: restore working state
signals.set_state("thinking")
signals.static_start()
if approved:
log("[perm] approved by voice")
return PermissionResultAllow(behavior="allow")
log(f"[perm] denied: {answer!r}")
return PermissionResultDeny(
behavior="deny",
message=f'Denied by voice. The user said: "{answer[:500]}"',
interrupt=False)
approved = _norm_speech(answer) in _YES
# the model keeps working either way: restore the working state
signals.set_state("thinking")
signals.static_start()
if approved:
log("[perm] approved by voice")
return PermissionResultAllow(behavior="allow")
log(f"[perm] denied: {answer!r}")
return PermissionResultDeny(
behavior="deny",
message=f'Denied by voice. The user said: "{answer[:500]}"',
interrupt=False)
return gate


Expand Down Expand Up @@ -892,11 +998,11 @@ async def handle(text: str, spoke_from: float | None = None) -> bool:
"confirm", "confirmed", "yes confirm",
"yes confirmed"):
verb = pend + ":confirmed"
elif not expired and not any(q in text.lower()
elif not expired and not any(q in _norm_speech(text)
for q in QUIT_PHRASES):
mouth.say("Staying as we are.")
return True
if any(q in text.lower() for q in QUIT_PHRASES):
if any(q in _norm_speech(text) for q in QUIT_PHRASES):
if speak_task and not speak_task.done():
speak_task.cancel()
mouth.shut_up()
Expand Down Expand Up @@ -1060,6 +1166,11 @@ async def handle(text: str, spoke_from: float | None = None) -> bool:
signals.set_state("idle")
await brain.stop()
log("[backtalk] hung up")
# our own cleanup above is done; asyncio.run()'s own shutdown
# (cancelling stray tasks, closing the default executor) can hang
# on native-library threads (torch/kokoro), so exit immediately
# here rather than let control return through that teardown.
os._exit(0)


# Loopback port used purely as a mutex. Nothing is ever served on it.
Expand Down Expand Up @@ -1111,6 +1222,7 @@ def main():
asyncio.run(amain())
except KeyboardInterrupt:
print("\n[backtalk] interrupted — hanging up", flush=True)
os._exit(0) # belt and suspenders: same reasoning as the exit above


if __name__ == "__main__":
Expand Down