Nostr 協議架構：去中心化社交網路的技術基礎

Nostr（Notes and Other Stuff Transmitted by Relays）是一個簡單而強大的協議，用於構建去中心化的社交網路和信息系統。與傳統社交平台不同，Nostr 不依賴任何中央伺服器，而是通過稱為「relay」的伺服器網路傳播信息。這個設計選擇使 Nostr 具有極強的抗審查能力和靈活性。

本文將深入分析 Nostr 的協議架構，包括客戶端、relay、密鑰體系和事件模型。我們將探討這個架構如何實現去中心化，以及開發者在構建 Nostr 應用時需要考慮的關鍵設計決策。

Nostr 的核心設計原則

極簡主義

Nostr 的設計哲學是「極簡」：

1. 極少的數據類型：只有一種基本數據結構 — 事件
2. 簡單的協議：只有幾種消息類型
3. 客戶端負責：relay 只做簡單的存儲和轉發，複雜邏輯交給客戶端

這種極簡主義的好處是：

• 協議容易被理解和實現
• 沒有複雜的共識機制
• 容易進行實驗和迭代

身份與密鑰

Nostr 採用密鑰對作為用戶身份：

┌─────────────────────────────────────────────────────────────┐
│                    Nostr 身份系統                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  用戶身份 = 一對密鑰                                        │
│  ┌─────────────────────┐                                   │
│  │  公鑰 (npub...)     │  類似「用戶名」                   │
│  │  私鑰 (nsec...)     │  類似「密碼」                    │
│  └─────────────────────┘                                   │
│                                                             │
│  優勢：                                                     │
│  - 無需註冊，直接使用                                       │
│  - 沒有中央機構可以撤銷帳戶                                 │
│  - 用戶完全控制自己的身份                                   │
│  - 可以創建多個帳戶（多個密鑰對）                           │
│                                                             │
└─────────────────────────────────────────────────────────────┘

中繼器（Relay）

Relay 是 Nostr 網路中的簡單伺服器：

• 接收客戶端發布的事件
• 存儲事件（可選，有存儲策略）
• 向訂閱的客戶端廣播事件
• 不執行任何業務邏輯

┌─────────────────────────────────────────────────────────────┐
│                    Nostr Relay 網路                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────┐   ┌─────────┐   ┌─────────┐                 │
│  │ Relay 1 │   │ Relay 2 │   │ Relay 3 │                 │
│  │ :8080   │   │ :8081   │   │ :8082   │                 │
│  └────┬────┘   └────┬────┘   └────┬────┘                 │
│       │             │             │                        │
│       └─────────────┼─────────────┘                        │
│                     │                                       │
│                     ▼                                       │
│            ┌────────────────┐                              │
│            │   客戶端連接    │                              │
│            │  多個 Relay    │                              │
│            └────────────────┘                              │
│                                                             │
│  用戶 A ───────────────────────────────────────────▶      │
│  用戶 B ───────────────────────────────────────────▶      │
│  用戶 C ───────────────────────────────────────────▶      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Nostr 事件模型

事件結構

Nostr 的核心數據結構是「事件」（Event）：

{
  "id": "<32-byte-hash>",
  "pubkey": "<32-byte-public-key>",
  "created_at": 1234567890,
  "kind": 1,
  "tags": [],
  "content": "Hello, Nostr!",
  "sig": "<64-byte-signature>"
}

每個字段的意義：

事件類型（Kind）

Nostr 定義了標準的事件類型：

┌─────────────────────────────────────────────────────────────┐
│                    Nostr 事件類型                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  kind: 0  → 設置元資料（名稱、头像等）                    │
│  kind: 1  → 短文本訊息（類似 Twitter 推文）               │
│  kind: 2  → 推薦 Relay                                    │
│  kind: 3  → 關注列表                                       │
│  kind: 4  → 加密 DM（舊版）                               │
│  kind: 5  → 刪除事件                                       │
│  kind: 6  → 點讚                                           │
│  kind: 7  → 轉發                                           │
│  kind: 8  → 閃電網路支付 invoice                          │
│  kind: 9  → 閃電網路支付請求                              │
│  kind: 10 → 加密 DM（新版）                                │
│  kind: 40 → 創建社群                                       │
│  kind: 41 → 加入社群                                       │
│                                                             │
│  自訂類型：                                                 │
│  kind: 30000 → Kind 30000 長期儲存（可替換）              │
│  kind: 30001 → Kind 30001 長期儲存（可追加）              │
│  kind: 30023 → Kind 30023 長文章                         │
│                                                             │
└─────────────────────────────────────────────────────────────┘

事件的創建與驗證

事件的創建流程：

import hashlib
import secp256k1
import json

class NostrEvent:
    def __init__(self, private_key, kind, content, tags=None):
        self.private_key = private_key
        self.pubkey = private_key.pubkey()
        self.kind = kind
        self.content = content
        self.tags = tags or []
        self.created_at = int(time.time())

    def sign(self):
        """對事件進行簽名"""
        # 1. 構建事件數據（不含 sig）
        event_data = {
            "pubkey": self.pubkey.hex(),
            "created_at": self.created_at,
            "kind": self.kind,
            "tags": self.tags,
            "content": self.content
        }

        # 2. 序列化（JSON 壓縮格式）
        serialized = json.dumps(event_data, separators=(',', ':'))

        # 3. 計算 SHA-256 哈希
        self.id = hashlib.sha256(serialized.encode()).hexdigest()

        # 4. 對哈希進行簽名
        self.sig = self.private_key.schnorr_sign(self.id)

        return self

    def to_dict(self):
        """轉換為標準字典格式"""
        return {
            "id": self.id,
            "pubkey": self.pubkey.hex(),
            "created_at": self.created_at,
            "kind": self.kind,
            "tags": self.tags,
            "content": self.content,
            "sig": self.sig.hex()
        }

事件驗證

客戶端和 relay 都應該驗證事件：

def verify_event(event):
    """驗證 Nostr 事件"""

    # 1. 驗證簽名
    message = event["id"]
    pubkey = event["pubkey"]
    sig = event["sig"]

    if not schnorr_verify(pubkey, message, sig):
        return False, "Invalid signature"

    # 2. 驗證時間戳（可選）
    current_time = int(time.time())
    if abs(event["created_at"] - current_time) > 86400 * 365:
        return False, "Timestamp too old or in future"

    # 3. 驗證內容長度
    if len(event["content"]) > 1000000:
        return False, "Content too long"

    # 4. 驗證 JSON 格式
    try:
        json.dumps(event)
    except:
        return False, "Invalid JSON"

    return True, "Valid"

Nostr 客戶端架構

客戶端職責

Nostr 客戶端承擔大部分的複雜邏輯：

1. 密鑰管理：生成和管理用戶密鑰
2. 事件處理：創建、簽名、發布事件
3. 數據聚合：從多個 relay 收集事件
4. 狀態管理：維護本地數據庫
5. 用戶界面：展示內容，處理用戶交互

多 Relay 連接

客戶端通常連接多個 relay 以確保數據可靠性：

class NostrClient:
    def __init__(self):
        self.relays = []
        self.subscriptions = {}
        self.event_cache = {}

    def add_relay(self, url, options=None):
        """添加 relay"""
        relay = NostrRelay(url, options or {})
        self.relays.append(relay)
        return relay

    async def subscribe(self, filters):
        """訂閱事件"""
        subscription_id = generate_id()

        # 對每個 relay 發送訂閱
        for relay in self.relays:
            await relay.subscribe(subscription_id, filters)

        self.subscriptions[subscription_id] = filters
        return subscription_id

    async def handle_message(self, message):
        """處理接收到的消息"""
        if message["type"] == "EVENT":
            event = message["event"]

            # 驗證事件
            valid, _ = verify_event(event)
            if not valid:
                return

            # 去重
            if event["id"] in self.event_cache:
                return

            # 存儲事件
            self.event_cache[event["id"]] = event

            # 更新 UI
            await self.update_ui(event)

訂閱過濾器

客戶端使用過濾器來訂閱感興趣的事件：

{
  "kinds": [1, 30023],
  "authors": ["<pubkey1>", "<pubkey2>"],
  "since": 1234567890,
  "until": 1234567900,
  "#t": ["bitcoin", "nostr"],
  "limit": 100
}

過濾器參數：

Nostr Relay 實現

簡單 Relay 實現

一個基本的 Nostr relay 只需要處理幾種消息類型：

import asyncio

class NostrRelay:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.events = {}  # event_id -> event
        self.subscriptions = {}  # subscription_id -> filter
        self.clients = set()

    async def handle_client(self, reader, writer):
        """處理客戶端連接"""
        client = NostrClientHandler(reader, writer, self)
        self.clients.add(client)

        try:
            await client.loop()
        finally:
            self.clients.remove(client)

    async def handle_message(self, client, message):
        """處理各類消息"""

        if message[0] == "EVENT":
            # 客戶端發布事件
            event = message[1]
            if await self.validate_event(event):
                self.events[event["id"]] = event
                await self.broadcast(event)

        elif message[0] == "REQ":
            # 客戶端訂閱
            sub_id = message[1]
            filters = message[2:]
            self.subscriptions[sub_id] = filters
            await self.send_matching_events(client, filters)

        elif message[0] == "CLOSE":
            # 客戶端取消訂閱
            sub_id = message[1]
            if sub_id in self.subscriptions:
                del self.subscriptions[sub_id]

    async def broadcast(self, event):
        """廣播事件給所有訂閱者"""
        for client in self.clients:
            for sub_id, filters in client.subscriptions.items():
                if self.match_filters(event, filters):
                    await client.send(["EVENT", sub_id, event])

Relay 存儲策略

Relay 可以選擇不同的存儲策略：

1. 全部存儲：存儲所有事件（適合小規模 relay）
2. 時間限制：只保留最近 N 天的事件
3. 類型限制：只保留特定 kind 的事件
4. 付費存儲：只存儲付費用戶的事件

class StorageStrategy:
    async def should_store(self, event):
        """判斷是否應該存儲事件"""
        pass

    async def cleanup(self):
        """清理過期事件"""
        pass

class TimeBasedStrategy(StorageStrategy):
    def __init__(self, retention_days=90):
        self.retention_days = retention_days
        self.cutoff = time.time() - (retention_days * 86400)

    async def should_store(self, event):
        return event["created_at"] > self.cutoff

    async def cleanup(self):
        # 刪除過期事件
        pass

Nostr 的最終一致性

挑戰

Nostr 的設計是「最終一致」的，這帶來一些挑戰：

1. 重複事件：同一事件可能從多個 relay 收到
2. 亂序到達：事件可能不按時間順序到達
3. Relay 離線：連接的 relay 可能暫時離線
4. 數據缺失：某些 relay 可能從未收到過某些事件

處理策略

客戶端必須處理這些最終一致性問題：

class NostrClientAdvanced:
    def __init__(self):
        self.events = {}  # id -> event
        self.author_events = {}  # pubkey -> [events]

    async def merge_event(self, event):
        """合併新事件"""
        event_id = event["id"]

        # 去重
        if event_id in self.events:
            return

        # 驗證
        valid, error = verify_event(event)
        if not valid:
            return

        # 存儲
        self.events[event_id] = event

        # 按作者索引
        author = event["pubkey"]
        if author not in self.author_events:
            self.author_events[author] = []
        self.author_events[author].append(event)

        # 按時間排序
        self.author_events[author].sort(
            key=lambda e: e["created_at"]
        )

    def get_events_for_feed(self, filters):
        """獲取時間線事件"""
        result = []

        for event in self.events.values():
            if self.match_filters(event, filters):
                result.append(event)

        # 按時間排序
        result.sort(key=lambda e: e["created_at"], reverse=True)

        return result[:filters.get("limit", 100)]

    def get_notes_count(self):
        """獲取尚未讀取的數量"""
        # 計算本地沒有顯示過的事件
        pass

實踐建議

Relay 選擇

選擇 relay 時應考慮：

1. 可靠性：運行時間穩定
2. 地理位置：延遲低
3. 存儲策略：符合需求
4. 審查政策：是否符合內容需求
5. 付費模式：是否收費

客戶端設計

構建 Nostr 客戶端時：

1. 多 relay 連接：至少連接 3-5 個 relay
2. 本地驗證：不要相信 relay 的數據
3. 離線支持：實現本地數據庫
4. 優化體驗：預加載、緩存、增量加載
5. 錯誤處理：優雅處理 relay 離線

安全考量

1. 私鑰保護：安全存儲私鑰
2. 事件驗證：驗證所有接收的事件
3. DM 加密：使用 Nostr 協議的加密 DM
4. 備份密鑰：安全備份助記詞

Nostr 與其他協議的比較

結論

Nostr 的架構體現了「簡單而強大」的設計哲學。通過將複雜性從協議轉移到客戶端，Nustr 實現了一個極簡但功能強大的去中心化社交協議。

對於開發者而言，理解 Nostr 的架構意味著：

1. 客戶端承擔大部分邏輯
2. Relay 是簡單的存儲和轉發節點
3. 最終一致性需要客戶端處理
4. 密鑰身份是核心設計

這種架構為構建抗審查的社交網路提供了堅實的基礎。隨著更多開發者和用戶採用 Nostr，我們可以期待看到更多創新的應用和工具。

相關資源

• Nostr 協議規範：https://github.com/nostr-protocol/nips
• Nostr 客戶端：Damus, Amethyst, Snort
• Nostr Relay：nostr-rs-relay, Flask Nostr