DEV Community

EME GUG
EME GUG

Posted on

Build a Shared Encrypted Scratchpad for Your Coding Agents (Python + AES-GCM, ~150 Lines)

Dạo này mình chạy song song 2-3 coding agent cho cùng một project: một con refactor backend, một con viết test, một con review. Vấn đề đến rất nhanh: chúng không biết nhau đang làm gì. Agent A đổi tên function, agent B vẫn viết test cho tên cũ. Cách tạm thời là bắt chúng ghi vào một file NOTES.md chung, nhưng file đó nằm ngay trong repo, dễ bị commit nhầm, và thường chứa những thứ không nên lộ như connection string staging, token tạm hay ghi chú về lỗ hổng chưa fix.

Trên HN gần đây có một Show HN về ý tưởng "shared encrypted scratchpad cho coding agents". Mình thấy ý tưởng hay nên tự build một bản tối giản để hiểu rõ phần lõi. Bài này chia sẻ lại cách làm: server chỉ giữ ciphertext, mọi việc mã hóa đều diễn ra ở client, có xử lý conflict khi nhiều agent ghi cùng lúc. Tổng cộng khoảng 150 dòng Python.

Kiến trúc: server "mù", client giữ key

Nguyên tắc quan trọng nhất là server không bao giờ thấy plaintext, kể cả tên note. Nếu server bị lộ (log, backup, bị ai đó đọc DB), thứ họ có chỉ là các blob base64 vô nghĩa.

graph LR
    A1[Agent: refactor] --> C1[pad CLI]
    A2[Agent: test] --> C2[pad CLI]
    A3[Agent: review] --> C3[pad CLI]
    C1 -->|ciphertext| S[(Scratchpad server<br/>SQLite)]
    C2 -->|ciphertext| S
    C3 -->|ciphertext| S
    K[Passphrase + salt] -.->|derive key| C1
    K -.-> C2
    K -.-> C3

Stack mình dùng:

  • Python 3.12
  • cryptography 43.x (AES-GCM, Scrypt)
  • FastAPI 0.115 + uvicorn cho server
  • httpx cho client
python3.12 -m venv .venv && source .venv/bin/activate
pip install "cryptography>=43" "fastapi>=0.115" uvicorn httpx
Enter fullscreen mode Exit fullscreen mode

Phần crypto: làm đúng ngay từ đầu

Đây là phần dễ làm sai nhất, nên mình giữ nó thật ngắn và chỉ dùng primitive có sẵn, không tự chế:

  • Scrypt để derive key từ passphrase. Derive ra 64 bytes rồi tách đôi: 32 bytes cho AES, 32 bytes cho việc hash tên note. Không dùng chung một key cho hai mục đích khác nhau.
  • AES-256-GCM với nonce random 12 bytes cho mỗi lần ghi. Tuyệt đối không reuse nonce với cùng key.
  • AAD (associated data) là note id. Nhờ vậy nếu ai đó trên server tráo blob của note plan sang note secrets, lúc decrypt sẽ fail ngay chứ không âm thầm trả về data sai.
# padcrypto.py
import os, base64, hashlib
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.scrypt import Scrypt

def derive_keys(passphrase: str, salt: bytes) -> tuple[bytes, bytes]:
    kdf = Scrypt(salt=salt, length=64, n=2**15, r=8, p=1)
    raw = kdf.derive(passphrase.encode())
    return raw[:32], raw[32:]  # (enc_key, id_key)

def note_id(id_key: bytes, name: str) -> str:
    # keyed hash: server không đoán được tên note từ id
    return hashlib.blake2b(name.encode(), key=id_key, digest_size=16).hexdigest()

def encrypt(enc_key: bytes, nid: str, plaintext: str) -> str:
    nonce = os.urandom(12)
    ct = AESGCM(enc_key).encrypt(nonce, plaintext.encode(), nid.encode())
    return base64.b64encode(nonce + ct).decode()

def decrypt(enc_key: bytes, nid: str, blob: str) -> str:
    raw = base64.b64decode(blob)
    return AESGCM(enc_key).decrypt(raw[:12], raw[12:], nid.encode()).decode()
Enter fullscreen mode Exit fullscreen mode

Salt không cần giữ bí mật, nhưng mỗi scratchpad nên có salt riêng. Mình generate một lần bằng python -c "import os;print(os.urandom(16).hex())" rồi lưu cùng config. Scrypt với n=2**15 mất khoảng 50-100ms trên laptop, đủ chậm để brute-force passphrase tốn kém nhưng không làm agent chờ lâu.

Server: lưu blob và chống ghi đè bằng version

Khi nhiều agent cùng ghi vào một note, kiểu lỗi kinh điển là lost update: agent A đọc, agent B đọc, A ghi, B ghi đè mất phần của A. Mình xử lý bằng optimistic concurrency, tức là mỗi note có version, client phải gửi kèm version đang thấy, lệch thì server trả 409.

# server.py
import sqlite3
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
db = sqlite3.connect("pad.db", check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS notes (id TEXT PRIMARY KEY, blob TEXT, version INTEGER)")

class Note(BaseModel):
    blob: str
    version: int  # version client đang thấy, 0 = note mới

@app.get("/notes/{nid}")
def get_note(nid: str):
    row = db.execute("SELECT blob, version FROM notes WHERE id=?", (nid,)).fetchone()
    if not row:
        raise HTTPException(404)
    return {"blob": row[0], "version": row[1]}

@app.put("/notes/{nid}")
def put_note(nid: str, note: Note):
    with db:
        if note.version == 0:
            cur = db.execute("INSERT OR IGNORE INTO notes VALUES (?, ?, 1)", (nid, note.blob))
        else:
            cur = db.execute(
                "UPDATE notes SET blob=?, version=version+1 WHERE id=? AND version=?",
                (note.blob, nid, note.version),
            )
        if cur.rowcount == 0:
            raise HTTPException(409, "version conflict")
    return {"version": note.version + 1}
Enter fullscreen mode Exit fullscreen mode

Chạy server chỉ bind localhost, đừng mở ra 0.0.0.0 nếu chưa có auth:

uvicorn server:app --host 127.0.0.1 --port 8787
Enter fullscreen mode Exit fullscreen mode

Luồng ghi khi có conflict trông như sau:

sequenceDiagram
    participant A as Agent A
    participant B as Agent B
    participant S as Server
    A->>S: GET plan (v3)
    B->>S: GET plan (v3)
    A->>S: PUT plan, version=3
    S-->>A: 200, v4
    B->>S: PUT plan, version=3
    S-->>B: 409 conflict
    B->>S: GET plan (v4), merge, PUT version=4
    S-->>B: 200, v5

Client CLI cho agent dùng

Agent làm việc tốt nhất với CLI đơn giản, đọc stdin và ghi stdout. Mình thêm lệnh append có tự retry khi gặp 409, vì 90% trường hợp agent chỉ muốn thêm một dòng log chứ không cần ghi đè cả note.

# pad.py
import os, sys, httpx
from padcrypto import derive_keys, note_id, encrypt, decrypt

URL = os.environ.get("PAD_URL", "http://127.0.0.1:8787")
ENC, IDK = derive_keys(os.environ["PAD_PASS"], bytes.fromhex(os.environ["PAD_SALT"]))

def fetch(nid):
    r = httpx.get(f"{URL}/notes/{nid}")
    if r.status_code == 404:
        return "", 0
    r.raise_for_status()
    d = r.json()
    return decrypt(ENC, nid, d["blob"]), d["version"]

def main(cmd, name):
    nid = note_id(IDK, name)
    if cmd == "get":
        print(fetch(nid)[0])
        return
    new = sys.stdin.read()
    for _ in range(5):  # retry khi conflict
        old, ver = fetch(nid)
        text = old + new if cmd == "append" else new
        r = httpx.put(f"{URL}/notes/{nid}", json={"blob": encrypt(ENC, nid, text), "version": ver})
        if r.status_code != 409:
            r.raise_for_status()
            return
    sys.exit("conflict: retry limit")

if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])
Enter fullscreen mode Exit fullscreen mode

Cách dùng thực tế:

export PAD_PASS="$(cat ~/.config/pad/pass)"
export PAD_SALT="$(cat ~/.config/pad/salt)"

echo "- [refactor] đổi get_user() -> fetch_user()" | python pad.py append changelog
python pad.py get changelog

# kiểm tra server thật sự không thấy gì
sqlite3 pad.db "SELECT id, substr(blob,1,40) FROM notes;"
Enter fullscreen mode Exit fullscreen mode

Cuối cùng, thêm vào file instruction của từng agent (kiểu AGENTS.md) một đoạn ngắn: trước khi bắt đầu task, chạy python pad.py get changelog; sau khi đổi API/interface, append một dòng mô tả. Sau khi thêm, mình thấy số lần agent viết code dựa trên interface cũ giảm hẳn.

Những thứ mình cố ý chưa làm (và bạn nên cân nhắc)

  • Auth cho server: hiện tại ai gọi được port 8787 đều ghi được blob rác (dù không đọc được nội dung). Nếu chạy trên máy dùng chung, thêm một bearer token đơn giản.
  • Rotate passphrase: đổi passphrase nghĩa là phải re-encrypt toàn bộ note. Với scratchpad tồn tại vài ngày thì xóa đi tạo lại là xong.
  • Prompt injection: encryption bảo vệ data khỏi server, không bảo vệ agent khỏi nội dung độc hại do agent khác ghi vào. Hãy coi nội dung scratchpad như input không tin cậy.
  • Passphrase trong env var: agent có quyền chạy shell thì đọc được env var. Đây là trade-off chấp nhận được với máy local, nhưng đừng dùng chung passphrase này cho thứ gì khác.

Kết luận

Shared scratchpad giải quyết đúng một vấn đề cụ thể là các agent chạy song song không biết nhau đang làm gì, và bản tối giản không cần nhiều code. Mấy điểm nên mang về:

  1. Mã hóa ở client, server chỉ lưu blob. Hash luôn cả tên note bằng keyed hash (BLAKE2b) để không lộ metadata.
  2. Dùng primitive chuẩn: Scrypt để derive key, AES-GCM với nonce random, note id làm AAD. Tách riêng key cho từng mục đích.
  3. Optimistic concurrency với version + HTTP 409 là cách rẻ nhất để chống lost update khi nhiều agent ghi cùng lúc.
  4. Thiết kế CLI theo kiểu append-first: agent chủ yếu cần ghi log, không cần ghi đè.
  5. Encryption không thay được trust boundary: vẫn coi nội dung do agent khác ghi là untrusted input.

Cuối tuần này bạn thử dựng bản này cho project đang chạy multi-agent, rồi xem log changelog sau vài ngày. Mình đoán bạn sẽ ngạc nhiên vì các agent đã đổi nhiều thứ mà bạn không hề biết.

Top comments (0)