grpc_log_repro.py

#!/usr/bin/env python3
"""Reproduce 'blocked gRPC host does not surface in sbx policy log'.

Run inside an sbx sandbox. Each approach targets a DIFFERENT, uniquely-named
(and therefore guaranteed-blocked) host, so afterwards you can look at
`sbx policy log <sandbox>` and see exactly which approach produced a log entry
and of what kind.

Observed on claude-test-connection (grpcio 1.82, no proxy CA trusted):

  A) plain local DNS lookup (getaddrinfo)   -> VISIBLE: `network <host>` (no port)
  B) gRPC WITH proxy, RPC issued            -> INVISIBLE: no local DNS (delegated
       to proxy via CONNECT), then dies at the proxy's MITM TLS handshake
       (untrusted self-signed proxy CA) before any host decision is logged.
  C) gRPC WITH proxy, channel only, NO RPC  -> INVISIBLE: lazy channel never dials
  D) gRPC with proxy DISABLED, RPC issued   -> VISIBLE: `network <host>` (proxy off
       forces local DNS, which the gateway DNS blocks and logs)

So a host lands in the log only when something does a LOCAL DNS lookup for it
(A, D) or completes a clean proxy CONNECT (e.g. curl -> `forward host:port`).
Plain gRPC-over-proxy (B, C) hides the host.

Each host is `<case>-<runid>.example.net` so they group together and are
easy to grep. Cases B and C are the "invisible" ones.

Usage:
  python3 grpc_log_repro.py [port]     # default port 8443
"""
import os
import socket
import sys
import time
import uuid

import grpc

PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8443
RUNID = uuid.uuid4().hex[:6]


def host(case: str) -> str:
    # Distinct, definitely-not-allow-listed host per approach.
    return f"{case}-{RUNID}.example.net"


HOSTS = {
    "A-dns": host("a-dns"),
    "B-grpc-proxy-rpc": host("b-grpc-proxy-rpc"),
    "C-grpc-proxy-norpc": host("c-grpc-proxy-norpc"),
    "D-grpc-noproxy-rpc": host("d-grpc-noproxy-rpc"),
}

print(f"grpc version: {grpc.__version__}")
print(f"run id      : {RUNID}   (port {PORT})")
print(f"http_proxy  : {os.environ.get('http_proxy') or os.environ.get('HTTP_PROXY')}")
print(f"grpc_proxy  : {os.environ.get('grpc_proxy')}")
print("=" * 68)


def _try_rpc(channel, tag):
    # Any unary call forces the channel to actually dial. The method need not
    # exist; we only care that a connection attempt is made.
    fn = channel.unary_unary("/repro.Svc/Ping")
    try:
        fn(b"", timeout=8)
        print(f"   {tag}: returned (unexpected)")
    except grpc.RpcError as e:
        print(f"   {tag}: RpcError {e.code()} - {e.details()}")
    except Exception as e:  # noqa: BLE001
        print(f"   {tag}: {type(e).__name__}: {e}")


def case_a_dns():
    h = HOSTS["A-dns"]
    print(f"A) socket.getaddrinfo (local DNS)          host={h}")
    try:
        socket.getaddrinfo(h, PORT, proto=socket.IPPROTO_TCP)
        print("   resolved (unexpected)")
    except socket.gaierror as e:
        print(f"   DNS failed: {e}")
    print(f"   -> expect `network {h}` in log")


def case_b_grpc_proxy_rpc():
    h = HOSTS["B-grpc-proxy-rpc"]
    print(f"B) gRPC WITH proxy, RPC issued             host={h}")
    ch = grpc.secure_channel(f"{h}:{PORT}", grpc.ssl_channel_credentials())
    _try_rpc(ch, "rpc")
    ch.close()
    print(f"   -> INVISIBLE: no local DNS; TLS to proxy MITM cert fails first")


def case_c_grpc_proxy_norpc():
    h = HOSTS["C-grpc-proxy-norpc"]
    print(f"C) gRPC WITH proxy, NO RPC (lazy channel)  host={h}")
    ch = grpc.secure_channel(f"{h}:{PORT}", grpc.ssl_channel_credentials())
    time.sleep(1)  # give it a chance to (not) do anything
    ch.close()
    print(f"   -> INVISIBLE for {h} (no DNS, no CONNECT)")


def case_d_grpc_noproxy_rpc():
    h = HOSTS["D-grpc-noproxy-rpc"]
    print(f"D) gRPC proxy DISABLED, RPC issued         host={h}")
    # enable_http_proxy=0 makes grpc resolve DNS itself instead of using CONNECT.
    ch = grpc.secure_channel(
        f"{h}:{PORT}",
        grpc.ssl_channel_credentials(),
        options=(("grpc.enable_http_proxy", 0),),
    )
    _try_rpc(ch, "rpc")
    ch.close()
    print(f"   -> expect `network {h}` in log (forced local DNS)")


if __name__ == "__main__":
    for fn in (case_a_dns, case_b_grpc_proxy_rpc, case_c_grpc_proxy_norpc, case_d_grpc_noproxy_rpc):
        fn()
        print("-" * 68)
    print("Hosts used this run:")
    for label, h in HOSTS.items():
        print(f"  {label:22} {h}")
    print("\nCheck the log on the HOST (not inside the sandbox):")
    print(f"  sbx policy log <sandbox> | grep {RUNID}")