書き込みより怖いのは到達先だった — エージェントの外向き通信を許可ホストだけに絞る

# egress_gate.py — エージェントの外向き通信を許可ホストだけに絞る deny-by-default ゲート
import asyncio, fnmatch, time
 
def load_allow(path="egress_allowlist.txt"):
    with open(path) as f:
        return [ln.strip() for ln in f if ln.strip() and not ln.startswith("#")]
 
ALLOW = load_allow()
LOG = open("egress_denied.log", "a")
 
def allowed(host: str) -> bool:
    # 完全一致か、*.example.com のワイルドカードで判定する
    return any(fnmatch.fnmatch(host, pat) for pat in ALLOW)
 
async def pipe(reader, writer):
    try:
        while data := await reader.read(65536):
            writer.write(data)
            await writer.drain()
    except Exception:
        pass
    finally:
        writer.close()
 
async def handle(client_r, client_w):
    try:
        head = await client_r.readuntil(b"\r\n\r\n")
    except Exception:
        client_w.close(); return
    line = head.split(b"\r\n", 1)[0].decode("latin1")
    parts = line.split(" ")
    # 到達先を秘匿するため CONNECT（HTTPS トンネル）のみ扱う
    if len(parts) < 2 or parts[0] != "CONNECT":
        client_w.write(b"HTTP/1.1 405 Method Not Allowed\r\n\r\n")
        await client_w.drain(); client_w.close(); return
    host, _, port = parts[1].partition(":")
    port = port or "443"
    if not allowed(host):
        LOG.write(f"{time.strftime('%F %T')}\tDENY\t{host}:{port}\n")
        LOG.flush()
        client_w.write(b"HTTP/1.1 403 Forbidden\r\n\r\n")
        await client_w.drain(); client_w.close(); return
    try:
        up_r, up_w = await asyncio.open_connection(host, int(port))
    except Exception:
        client_w.write(b"HTTP/1.1 502 Bad Gateway\r\n\r\n")
        await client_w.drain(); client_w.close(); return
    client_w.write(b"HTTP/1.1 200 Connection Established\r\n\r\n")
    await client_w.drain()
    await asyncio.gather(pipe(client_r, up_w), pipe(up_r, client_w))
 
async def main():
    server = await asyncio.start_server(handle, "127.0.0.1", 8899)
    print("egress gate on 127.0.0.1:8899 / allow:", ALLOW)
    async with server:
        await server.serve_forever()
 
asyncio.run(main())

export HTTPS_PROXY="http://127.0.0.1:8899"
export https_proxy="http://127.0.0.1:8899"
# HTTP を使うツール対策（多くは 443 に寄せたいので必要最小限に）
export HTTP_PROXY="http://127.0.0.1:8899"

# egress_allowlist.txt — 到達を許すホストだけを1行ずつ
# コード取得・パッケージ
github.com
*.githubusercontent.com
registry.npmjs.org
# 使っている生成 API
generativelanguage.googleapis.com
# 自分のアプリ運用 API（鍵を持つ先だからこそ明示する）
androidpublisher.googleapis.com