Zero-Trust Signal Gateways for Local AI Agents: Hardening JSON-RPC Envelopes, Attachment Quarantine, and Execution Boundaries

In Supercharging Local AI Agents on Signal: Fast-Paths, Live Typing Indicators, and Session Continuity, we turned a local CLI coding assistant (Google Antigravity (agy) / Claude Code) into a sub-second mobile command center over Signal. By combining deterministic fast-paths, real-time sendTyping pulses over signal-cli JSON-RPC, and sliding 2-hour session continuity, we eliminated the 40-second mobile latency bottleneck.

Then, as The Autonomous Homelab cognitive architecture evolved—adding Out-of-Band Memory Dreaming, Procedural Skill Synthesis and Mistake Ledgers, and automatic Teacher–Student curriculum ingestion directly from Signal file attachments—the stakes changed dramatically.

Stop and think about what a headless AI agent wired to a messaging daemon actually is:

When you invoke a local coding agent with --dangerously-skip-permissions behind a chat listener, your messaging bot is no longer a chatbot. It is an autonomous remote shell with full user privileges, SSH key access, and infrastructure control.

If a single external message, a shared Signal group member, a malformed JSON-RPC envelope, a crafted attachment filename, or an unprivileged local Linux process can slip an instruction past your gateway boundary, they do not merely get a funny LLM reply—they gain arbitrary command execution on your machine.

To guarantee that only the authenticated owner can ever issue instructions and that every boundary holds under adversarial conditions, I conducted a full line-by-line security audit of signal_bot.py and its underlying Linux runtime. In this post, I break down the five subtle vulnerability classes uncovered during that audit and how to engineer a 5-layer Zero-Trust Signal Gateway.

---

1. The 5-Layer Defense-in-Depth Architecture

A resilient agent gateway cannot rely on a single if sender == MY_NUMBER check. Every stage—from OS process IPC and JSON-RPC envelope parsing to file attachment staging, thread concurrency, and POSIX filesystem modes—must enforce its own boundary invariants:

Let’s examine each layer, the specific pitfalls that standard implementations miss, and the exact code that closes them.

---

2. Layer 1: Process & Transport Isolation (stdio Pipes vs. TCP/HTTP Daemons)

The Vulnerability: Unauthenticated Local Sockets

Most signal-cli integration guides recommend starting the daemon with a local TCP or HTTP listener:

# ANTI-PATTERN: Exposes an unauthenticated JSON-RPC socket on localhost!
signal-cli -u +15550199000 daemon --tcp 127.0.0.1:7583

On a developer machine or server, binding an unauthenticated JSON-RPC daemon to 127.0.0.1 is a severe local privilege-escalation and SSRF vector:

1. Cross-Process Injection: Any container sharing the host network namespace, any unprivileged local service, or any compromised npm/PyPI test runner can connect to 127.0.0.1:7583, read every incoming Signal message, or inject synthetic events.
2. Browser / Webhook SSRF: Local web services or browser extensions vulnerable to Server-Side Request Forgery can hit localhost ports to exfiltrate messages or trigger actions.

The Fix: Anonymous Kernel stdio Pipes & 0700 Keystores

Instead of running a network listener, signal_bot.py spawns signal-cli -u <ACCOUNT> jsonRpc directly as a child process connected exclusively via anonymous kernel pipes (stdin=subprocess.PIPE, stdout=subprocess.PIPE):

proc = subprocess.Popen(
    ["signal-cli", "-u", ALLOWED_ACCOUNT, "jsonRpc"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    bufsize=1,
)

Because communication happens strictly over unshared file descriptors between parent (python3) and child (java), no socket is ever bound. We verify this live on the host:

$ ss -tlpn | grep -E "java|python|signal" || echo "No listening TCP sockets"
No listening TCP sockets

$ stat -c "%a %n" ~/.local/share/signal-cli ~/.local/share/signal-cli/data/accounts.json
700 /home/dev/.local/share/signal-cli
600 /home/dev/.local/share/signal-cli/data/accounts.json

---

3. Layer 2: Outer Envelope Perimeter Gate & Defeating the or-Short-Circuit Trap

When you use your own Signal account’s “Note to Self” chat as your private AI command channel, understanding how signal-cli structures JSON-RPC envelopes is critical.

How signal-cli Delivers Messages

Your linked signal-cli daemon receives two fundamentally different message structures:

1. dataMessage: Direct messages sent to you by external contacts or members of any Signal group you belong to, where the sender sits in envelope.sourceNumber and envelope.sourceUuid.
2. syncMessage.sentMessage: Whenever you send a message from your phone or Signal Desktop—whether to “Note to Self”, to a friend, or to a 50-person group chat—your phone sends a syncMessage to all of your linked devices so your chat history stays in sync. Inside sentMessage, the recipient sits in destinationNumber / destinationUuid.

Two Subtle Identity-Check Bugs

In the first version of our gateway, the envelope loop contained two subtle logic traps that looked secure at a glance:

# BUG 1: Short-circuit `or` only inspects the first truthy field!
source = env.get("sourceNumber") or env.get("sourceUuid")

# BUG 2: Entering `syncMessage` and checking `dest` BEFORE verifying outer `source`!
if "syncMessage" in env and "sentMessage" in env["syncMessage"]:
    sent = env["syncMessage"]["sentMessage"]
    dest = sent.get("destinationNumber") or sent.get("destinationUuid")
    if dest in ALLOWED_IDS:
        # Executes prompt!

Look closely at why this fails strict boundary rigor:

1. The or-Short-Circuit Trap: In Python, a or b evaluates to a whenever a is non-empty, completely ignoring b. If a malformed or adversarial envelope ever presented sourceNumber == "+15550199000" alongside a mismatched sourceUuid == "00000000-0000-0000-0000-000000000000" (or vice versa on destinationNumber / destinationUuid), source in ALLOWED_IDS evaluates to True!
2. Missing Outer-Envelope Verification on syncMessage: While the Signal protocol normally generates syncMessage envelopes only from your linked devices, trusting an inner syncMessage.sentMessage.destination field without first proving at the outer envelope perimeter that the sender (sourceNumber + sourceUuid) is you violates defense-in-depth.

The Fix: Strict Dual-Field Conjunction (is_authorized_self_identity)

To make illegal identity states unrepresentable, we replaced loose or checks with a strict conjunction validator:

ALLOWED_ACCOUNT = "+15550199000"
ALLOWED_UUID = "a1b2c3d4-e5f6-4a5b-8c9d-0123456789ab"
ALLOWED_IDS = {ALLOWED_ACCOUNT, ALLOWED_UUID}


def is_authorized_self_identity(number: str | None, uuid: str | None) -> bool:
    """Strictly verify that number and/or uuid belong exclusively to the owner."""
    num = (number or "").strip() or None
    uid = (uuid or "").strip() or None
    if not num and not uid:
        return False
    if num is not None and num != ALLOWED_ACCOUNT:
        return False
    if uid is not None and uid != ALLOWED_UUID:
        return False
    return True

Notice the properties of is_authorized_self_identity():

• If both number and uuid are present, both must match ALLOWED_ACCOUNT and ALLOWED_UUID respectively.
• If either field is present and mismatches, the function immediately returns False.
• If neither is present ((None, None) or ("", "")), it returns False.

We then enforce this check at two separate gates—first at the outer envelope boundary before inspecting any inner payload, and again on the inner destination of syncMessage:

src_num = env.get("sourceNumber") or env.get("source")
src_uuid = env.get("sourceUuid")

# Gate 1: Reject any group or story context at the outer envelope
if "groupInfo" in env or "groupContext" in env or "groupId" in env or "storyMessage" in env:
    continue

# Gate 2: Enforce strict owner identity at the outer envelope perimeter
if not is_authorized_self_identity(src_num, src_uuid):
    if "dataMessage" in env or "syncMessage" in env:
        print(f"SECURITY BLOCKED: Ignored envelope from non-owner ({src_num} / {src_uuid})", flush=True)
    continue

# Gate 3: For syncMessage, destination must ALSO be strictly self (Note to Self) and non-group
if "syncMessage" in env and "sentMessage" in env["syncMessage"]:
    sent = env["syncMessage"]["sentMessage"]
    dest_num = sent.get("destinationNumber") or sent.get("destination")
    dest_uuid = sent.get("destinationUuid")
    group_info = sent.get("groupInfo") or sent.get("groupContext") or sent.get("group") or sent.get("groupId")
    if group_info or not is_authorized_self_identity(dest_num, dest_uuid):
        continue

We prove the gate with a 9-case assertion battery covering every positive, empty, external, and split-field spoofing combination:

assert is_authorized_self_identity(ALLOWED_ACCOUNT, ALLOWED_UUID) is True
assert is_authorized_self_identity(ALLOWED_ACCOUNT, None) is True
assert is_authorized_self_identity(None, ALLOWED_UUID) is True
assert is_authorized_self_identity(None, None) is False
assert is_authorized_self_identity("", "") is False
assert is_authorized_self_identity("+15550000000", None) is False
assert is_authorized_self_identity(None, "00000000-0000-0000-0000-000000000000") is False
assert is_authorized_self_identity(ALLOWED_ACCOUNT, "00000000-0000-0000-0000-000000000000") is False
assert is_authorized_self_identity("+15550000000", ALLOWED_UUID) is False

---

4. Layer 3: Attachment Path Traversal Containment & Supply-Chain Quarantine

To support dropping teacher-curriculum-v1 JSON bundles directly into Signal’s “Note to Self” chat, signal_bot.py inspects incoming file attachments.

The Vulnerabilities

1. Path Traversal via attachment.id or bundle_id:
   Naively constructing paths from JSON fields such as:
   

   # VULNERABLE: What if att_id or bundle_id contains "../../"?
   src = SIGNAL_ATTACHMENTS_DIR / str(att["id"])
   dst = TEACHER_INBOX_DIR / f"teacher-curriculum-{date_part}-{data['bundle_id']}.json"

   
   allows crafted identifiers containing ../ to read arbitrary files outside ~/.local/share/signal-cli/attachments or write files outside 00-Inbox/teacher/.
2. Auto-Installing Executable SKILL.md Runbooks:
   A curriculum bundle can carry procedural SKILL.md files containing shell commands that the agent will trust on future runs. Auto-installing skills from a chat attachment without human review creates a direct supply-chain persistence vector.

The Fix: Regex Allowlists, Canonical .resolve().parent Verification & Mandatory Quarantine

We enforce strict allowlist regexes (SAFE_FILENAME_RE, SAFE_DATE_RE, SAFE_BUNDLE_ID_RE), verify that the resolved canonical parent directory matches the expected root, cap file size at 2 MB, and force --skill-mode review:

SAFE_FILENAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,128}$")
SAFE_BUNDLE_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,16}$")
SAFE_DATE_RE = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}$")


def process_curriculum_attachments(rpc_proc, attachments: list, msg_text: str | None) -> bool:
    if not attachments:
        return False
    staged = []
    TEACHER_INBOX_DIR.mkdir(parents=True, exist_ok=True)
    inbox_resolved = TEACHER_INBOX_DIR.resolve()
    att_dir_resolved = SIGNAL_ATTACHMENTS_DIR.resolve()

    for att in attachments:
        if not isinstance(att, dict):
            continue
        att_id = str(att.get("id") or "").strip()
        if not att_id or not SAFE_FILENAME_RE.match(att_id):
            continue
        src = (SIGNAL_ATTACHMENTS_DIR / att_id).resolve()
        if src.parent != att_dir_resolved or not src.is_file() or src.stat().st_size > 2_000_000:
            continue
        try:
            data = json.loads(src.read_text(encoding="utf-8"))
        except Exception:
            continue
        if isinstance(data, dict) and data.get("schema") == "teacher-curriculum-v1" and "bundle_id" in data:
            raw_date = str(data.get("created_at") or time.strftime("%Y-%m-%d"))[:10]
            date_part = raw_date if SAFE_DATE_RE.match(raw_date) else time.strftime("%Y-%m-%d")
            raw_bid = re.sub(r"[^A-Za-z0-9_-]", "", str(data["bundle_id"]))[:12]
            if not raw_bid or not SAFE_BUNDLE_ID_RE.match(raw_bid):
                continue
            dst = (TEACHER_INBOX_DIR / f"teacher-curriculum-{date_part}-{raw_bid}.json").resolve()
            if dst.parent != inbox_resolved:
                continue
            if not dst.exists():
                dst.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
            staged.append(dst.name)

    if not staged and not (msg_text and "[TEACHER_CURRICULUM]" in msg_text):
        return False

    # Enforce --skill-mode review so executable SKILL.md files are quarantined first
    proc = subprocess.run(
        [sys.executable, str(SCRIPTS_DIR / "student_assimilate.py"), "--skill-mode", "review"],
        capture_output=True, text=True, timeout=60,
    )
    ...

Any incoming skill is placed into 00-Inbox/teacher/skills-review/<skill-name>/SKILL.md where it remains inert until the owner inspects it and runs --promote-skill.

---

5. Layer 4: Execution Serialization, Echo-Loop Guards & Model Flag Resilience

Even when every message comes from you, concurrency and runtime state changes can destabilize a headless agent daemon.

1. Serializing --conversation Execution (AGY_EXEC_LOCK)

When signal_bot.py receives a message, it spawns a worker thread so the main JSON-RPC reader loop never blocks. However, if you send two complex prompts 3 seconds apart on your phone, two concurrent threads will read the same active_cid from signal_conversation.json and spawn two simultaneous agy --conversation <cid> processes racing on the same transcript and state file!

Wrapping the AI execution branch in a dedicated mutex (AGY_EXEC_LOCK) while leaving deterministic fast-path commands (ping, status, weather) outside the lock gives you the best of both worlds—instant fast-paths even while a deep agent turn is running, and zero concurrent conversation corruption:

AGY_EXEC_LOCK = threading.Lock()

# Inside process_message() after fast-path check:
try:
    with AGY_EXEC_LOCK:
        active_cid = get_active_conversation_id()
        reply, new_cid = execute_agy(prompt, active_cid=active_cid)
        if new_cid:
            save_active_conversation_id(new_cid)
finally:
    pulser.stop()

2. Breaking syncMessage Self-Reply Loops (RECENT_REPLIES)

When the bot sends a reply back to your own number, signal-cli may echo that outbound text as an incoming syncMessage. If the bot treats its own reply as a new user prompt, it will enter a runaway self-conversation loop burning tokens until rate-limited.

Maintaining a bounded ring buffer (RECENT_REPLIES = collections.deque(maxlen=32)) populated inside send_signal_reply() and checked before dispatching any message stops self-reply echoes in O(1) time.

3. Self-Healing Model Flag Probing (_EFFORT_SUPPORTED) & Graceful Timeout Containment (_run_agy_cmd)

Two runtime traps surface when driving a CLI coding agent headlessly via subprocess.run():

1. Model Flag Rejection (--effort): To keep mobile responses snappy, signal_bot.py passes --effort low to agy. However, if you switch your active workspace model from a thinking/reasoning model to a non-thinking model, the CLI exits with rc=1 (--effort is not supported for the current model).
2. Subprocess Timeout Race & sys.argv Exception Leakage: If Python’s subprocess.run(..., timeout=300) shares the exact same 300-second deadline as --print-timeout 5m, Python kills the child process at the boundary and raises subprocess.TimeoutExpired. If uncaught inside execute_agy(), str(e) formats the entire cmd array—leaking internal binary paths, CLI flags, and hidden system-prompt wrappers directly into the chat message while losing the conversation_id needed to resume the multi-step task!

We solve both by wrapping execution in _run_agy_cmd() with a 10m agent print timeout (630s Python grace window), catching subprocess.TimeoutExpired cleanly, preserving the latest conversation_id on timeout so the user can reply continue, and dynamically probing _EFFORT_SUPPORTED:

_EFFORT_SUPPORTED: bool | None = None
AGY_PRINT_TIMEOUT = "10m"
AGY_SUBPROCESS_TIMEOUT = 630


def _run_agy_cmd(cmd: list[str]) -> tuple[int, str, str, bool]:
    """Run agy subprocess safely, catching TimeoutExpired without leaking cmd argv."""
    try:
        proc = subprocess.run(
            cmd,
            stdin=subprocess.DEVNULL,
            capture_output=True,
            text=True,
            timeout=AGY_SUBPROCESS_TIMEOUT,
        )
        return proc.returncode, (proc.stdout or "").strip(), (proc.stderr or "").strip(), False
    except subprocess.TimeoutExpired as e:
        out = (e.stdout.decode("utf-8", errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or "")).strip()
        err = (e.stderr.decode("utf-8", errors="replace") if isinstance(e.stderr, bytes) else (e.stderr or "")).strip()
        return 124, out, err, True


def execute_agy(prompt: str, active_cid: str = None) -> tuple[str, str]:
    global _EFFORT_SUPPORTED
    cmd = [AGY_BIN, "--dangerously-skip-permissions"]
    if _EFFORT_SUPPORTED is not False:
        cmd.extend(["--effort", "low"])
    cmd.extend(["--mode", "auto", "--print-timeout", AGY_PRINT_TIMEOUT])
    if active_cid:
        cmd.extend(["--conversation", active_cid])
    cmd.extend(["-p", prompt])

    rc, out, err, timed_out = _run_agy_cmd(cmd)
    if not timed_out and rc != 0 and "--effort is not supported" in err:
        _EFFORT_SUPPORTED = False
        cmd_no_effort =  != "--effort")]
        rc, out, err, timed_out = _run_agy_cmd(cmd_no_effort)
    elif not timed_out and rc == 0 and _EFFORT_SUPPORTED is None and "--effort" in cmd:
        _EFFORT_SUPPORTED = True

    new_cid = get_latest_conversation_id()
    if timed_out:
        partial = f"\n\nPartial output:\n{out[:1500]}" if out else ""
        reply = (
            f"⏱️ Task reached the {AGY_PRINT_TIMEOUT} execution limit before finishing the final step. "
            f"Session state ({new_cid[:8] if new_cid else 'active'}) is saved — reply `continue` to resume.{partial}"
        )
        return reply, new_cid
    return (out if out else err), new_cid

---

6. Layer 5: Closing the Linux $HOME o+x Traversal Gap

The final finding from our security audit is a classic Linux filesystem trap that catches many engineers off guard.

On modern Linux distributions (especially when ACLs or local media sharing are enabled), check the permissions on your home directory:

$ ls -ld $HOME
drwx--x---+ 92 dev dev 12288 Nov 14 10:00 /home/dev

Notice that trailing x (o+x / ACL execute bit). While o-r prevents another local user or compromised service account from listing /home/dev, o+x allows any local UID to traverse through /home/dev to open any file whose path they know or can guess, provided that file is world-readable (0644 or 0755)!

Prior to hardening:

• ~/scripts/signal_bot.py was 0755 (-rwxr-xr-x), exposing the allowed account number and UUID to any local user.
• ~/.gemini/antigravity-cli/signal_conversation.json was created with Python’s default umask (0644), allowing any local process to read the active conversation_id.
• Auxiliary API token files in ~/scripts/ were 0644.

The Fix: Atomic 0600 State Swaps & 0700 Script Lockdown

First, we updated save_active_conversation_id() so that conversation state is always written to a temporary file, locked down to 0600 (-rw-------) before being atomically renamed into place:

def save_active_conversation_id(cid):
    if not cid:
        return
    CONV_STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    payload = {"conversation_id": cid, "last_activity": time.time()}
    tmp_path = CONV_STATE_FILE.with_suffix(".tmp")
    tmp_path.write_text(json.dumps(payload), encoding="utf-8")
    os.chmod(tmp_path, 0o600)
    os.replace(tmp_path, CONV_STATE_FILE)

Second, we locked down the gateway script and all token/state files to owner-only access (700 / 600).

---

7. Complete Verification Audit Script

Never assume a security control works—prove it against the live system. You can run this single verification one-liner after deploying or updating your Signal gateway:

# 1. Verify zero listening TCP sockets from signal-cli or python
ss -tlpn 2>/dev/null | grep -E "java|python|signal" || echo "[PASS] 0 listening TCP sockets"

# 2. Verify syntax and run the 9 adversarial identity assertions
python3 -m py_compile ~/scripts/signal_bot.py && python3 -c '
import sys, pathlib
sys.path.insert(0, str(pathlib.Path.home() / "scripts"))
import signal_bot as sb

assert sb.is_authorized_self_identity(sb.ALLOWED_ACCOUNT, sb.ALLOWED_UUID) is True
assert sb.is_authorized_self_identity(sb.ALLOWED_ACCOUNT, None) is True
assert sb.is_authorized_self_identity(None, sb.ALLOWED_UUID) is True
assert sb.is_authorized_self_identity(None, None) is False
assert sb.is_authorized_self_identity("", "") is False
assert sb.is_authorized_self_identity("+15550000000", None) is False
assert sb.is_authorized_self_identity(None, "00000000-0000-0000-0000-000000000000") is False
assert sb.is_authorized_self_identity(sb.ALLOWED_ACCOUNT, "00000000-0000-0000-0000-000000000000") is False
assert sb.is_authorized_self_identity("+15550000000", sb.ALLOWED_UUID) is False
print("[PASS] 9/9 identity boundary assertions verified")
'

# 3. Verify owner-only file permissions (700 / 600)
stat -c "%a %n" \
  ~/scripts/signal_bot.py \
  ~/.gemini/antigravity-cli/signal_conversation.json \
  ~/.local/share/signal-cli \
  ~/.local/share/signal-cli/data/accounts.json

Expected output:

[PASS] 0 listening TCP sockets
[PASS] 9/9 identity boundary assertions verified
700 /home/dev/scripts/signal_bot.py
600 /home/dev/.gemini/antigravity-cli/signal_conversation.json
700 /home/dev/.local/share/signal-cli
600 /home/dev/.local/share/signal-cli/data/accounts.json

---

Conclusion

Connecting a local AI agent to Signal gives you a superpower in your pocket—but only if the gateway treats every incoming byte, envelope field, attachment filename, and local file descriptor with strict zero-trust discipline.

By isolating signal-cli over anonymous stdio pipes, enforcing dual-field (number, uuid) identity conjunction at the outer envelope perimeter, quarantining incoming attachments and skills, serializing agent execution under a lock, and sealing local file permissions against $HOME traversal, you can command your infrastructure from anywhere with confidence that only you hold the keys.