# TODO Analysis: Noise Security Issues

[acul71]

TODO Analysis: Noise Security Issues

Last review: August 2025

Overview

This document analyzes two TODO/FIXME issues in the noise security module:

1. Message prefix behavior in Go - libp2p/security/noise/io.py:44 DONE
2. Static key storage for IK pattern - libp2p/security/noise/transport.py:35 DEPRECATED in specs https://github.com/libp2p/specs/tree/master/noise#handshake-pattern

---

Issue 1: Message Prefix Behavior in Go

DONE
NOTE: Not in use but can be selected via a parameter

---

Issue 2: Static Key Storage for IK Pattern

Issue Location

File: libp2p/security/noise/transport.py:35
Class: Transport
TODO Comment: # TODO: A storage of seen noise static keys for pattern IK?

Current Status Analysis

Current Implementation

The Transport class currently doesn't implement noise pipes or IK pattern support:

class Transport(ISecureTransport):
    libp2p_privkey: PrivateKey
    noise_privkey: PrivateKey
    local_peer: ID
    early_data: bytes | None
    with_noise_pipes: bool

    # NOTE: Implementations that support Noise Pipes must decide whether to use
    #   an XX or IK handshake based on whether they possess a cached static
    #   Noise key for the remote peer.
    # TODO: A storage of seen noise static keys for pattern IK?

    def __init__(
        self,
        libp2p_keypair: KeyPair,
        noise_privkey: PrivateKey,
        early_data: bytes | None = None,
        with_noise_pipes: bool = False,
    ) -> None:
        self.libp2p_privkey = libp2p_keypair.private_key
        self.noise_privkey = noise_privkey
        self.local_peer = ID.from_pubkey(libp2p_keypair.public_key)
        self.early_data = early_data
        self.with_noise_pipes = with_noise_pipes

        if self.with_noise_pipes:
            raise NotImplementedError

    def get_pattern(self) -> IPattern:
        if self.with_noise_pipes:
            raise NotImplementedError
        else:
            return PatternXX(
                self.local_peer,
                self.libp2p_privkey,
                self.noise_privkey,
                self.early_data,
            )

The Problem

• Noise Pipes Not Implemented: The feature is disabled with NotImplementedError
• Missing IK Pattern: No implementation of the IK handshake pattern
• No Key Caching: No storage mechanism for static keys
• Performance Impact: Can't optimize handshakes with known peers

What Must Be Done

Required Changes

1. Static Key Storage: Implement a cache for remote peer static keys
2. IK Pattern Implementation: Create PatternIK class
3. Pattern Selection Logic: Choose between XX and IK based on cached keys
4. Key Management: Handle key storage, retrieval, and expiration

Implementation Options

Option A: Simple In-Memory Key Cache

import time
from typing import Dict, Optional

class Transport(ISecureTransport):
    libp2p_privkey: PrivateKey
    noise_privkey: PrivateKey
    local_peer: ID
    early_data: bytes | None
    with_noise_pipes: bool
    _static_key_cache: Dict[ID, tuple[PublicKey, float]]  # peer_id -> (key, timestamp)

    def __init__(
        self,
        libp2p_keypair: KeyPair,
        noise_privkey: PrivateKey,
        early_data: bytes | None = None,
        with_noise_pipes: bool = False,
        cache_ttl: int = 3600,  # 1 hour default
    ) -> None:
        self.libp2p_privkey = libp2p_keypair.private_key
        self.noise_privkey = noise_privkey
        self.local_peer = ID.from_pubkey(libp2p_keypair.public_key)
        self.early_data = early_data
        self.with_noise_pipes = with_noise_pipes
        self.cache_ttl = cache_ttl
        self._static_key_cache = {}

    def _get_cached_static_key(self, peer_id: ID) -> Optional[PublicKey]:
        """Get cached static key for a peer if available and not expired."""
        if peer_id not in self._static_key_cache:
            return None
        
        key, timestamp = self._static_key_cache[peer_id]
        if time.time() - timestamp > self.cache_ttl:
            del self._static_key_cache[peer_id]
            return None
        
        return key

    def _cache_static_key(self, peer_id: ID, static_key: PublicKey) -> None:
        """Cache a static key for a peer."""
        self._static_key_cache[peer_id] = (static_key, time.time())

    def get_pattern(self, remote_peer: ID | None = None) -> IPattern:
        if not self.with_noise_pipes:
            return PatternXX(
                self.local_peer,
                self.libp2p_privkey,
                self.noise_privkey,
                self.early_data,
            )
        
        # Check if we have a cached static key for IK pattern
        if remote_peer and self._get_cached_static_key(remote_peer):
            return PatternIK(
                self.local_peer,
                self.libp2p_privkey,
                self.noise_privkey,
                self.early_data,
                remote_peer,
                self._get_cached_static_key(remote_peer),
            )
        else:
            return PatternXX(
                self.local_peer,
                self.libp2p_privkey,
                self.noise_privkey,
                self.early_data,
            )

    async def secure_outbound(self, conn: IRawConnection, peer_id: ID) -> ISecureConn:
        pattern = self.get_pattern(peer_id)
        secure_conn = await pattern.handshake_outbound(conn, peer_id)
        
        # Cache the static key if using XX pattern (we learned it)
        if isinstance(pattern, PatternXX):
            # Extract static key from the handshake
            static_key = self._extract_static_key_from_handshake(pattern)
            if static_key:
                self._cache_static_key(peer_id, static_key)
        
        return secure_conn

Option B: Persistent Key Storage

import sqlite3
import json
from typing import Optional

class StaticKeyStorage:
    def __init__(self, db_path: str = "noise_static_keys.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self) -> None:
        """Initialize the database schema."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS static_keys (
                    peer_id TEXT PRIMARY KEY,
                    static_key BLOB,
                    created_at INTEGER,
                    last_used INTEGER,
                    use_count INTEGER DEFAULT 0
                )
            """)
    
    def get_static_key(self, peer_id: ID) -> Optional[PublicKey]:
        """Get cached static key for a peer."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT static_key, created_at FROM static_keys 
                WHERE peer_id = ?
            """, (str(peer_id),))
            
            row = cursor.fetchone()
            if row:
                static_key_bytes, created_at = row
                # Check if key is not too old (e.g., 24 hours)
                if time.time() - created_at < 86400:
                    # Update usage statistics
                    conn.execute("""
                        UPDATE static_keys 
                        SET last_used = ?, use_count = use_count + 1
                        WHERE peer_id = ?
                    """, (int(time.time()), str(peer_id)))
                    
                    return self._deserialize_public_key(static_key_bytes)
        
        return None
    
    def store_static_key(self, peer_id: ID, static_key: PublicKey) -> None:
        """Store a static key for a peer."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT OR REPLACE INTO static_keys 
                (peer_id, static_key, created_at, last_used)
                VALUES (?, ?, ?, ?)
            """, (
                str(peer_id),
                static_key.to_bytes(),
                int(time.time()),
                int(time.time())
            ))
    
    def cleanup_expired_keys(self, max_age: int = 86400) -> None:
        """Remove expired keys from storage."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                DELETE FROM static_keys 
                WHERE ? - created_at > ?
            """, (int(time.time()), max_age))

class Transport(ISecureTransport):
    def __init__(
        self,
        libp2p_keypair: KeyPair,
        noise_privkey: PrivateKey,
        early_data: bytes | None = None,
        with_noise_pipes: bool = False,
        key_storage: StaticKeyStorage | None = None,
    ) -> None:
        # ... existing initialization ...
        self.key_storage = key_storage or StaticKeyStorage()
        self.with_noise_pipes = with_noise_pipes

    def get_pattern(self, remote_peer: ID | None = None) -> IPattern:
        if not self.with_noise_pipes:
            return PatternXX(...)
        
        # Check persistent storage for static key
        if remote_peer:
            cached_key = self.key_storage.get_static_key(remote_peer)
            if cached_key:
                return PatternIK(
                    self.local_peer,
                    self.libp2p_privkey,
                    self.noise_privkey,
                    self.early_data,
                    remote_peer,
                    cached_key,
                )
        
        return PatternXX(...)

Option C: Advanced PatternIK Implementation

class PatternIK(BasePattern):
    """Noise IK pattern implementation for noise pipes."""
    
    def __init__(
        self,
        local_peer: ID,
        libp2p_privkey: PrivateKey,
        noise_static_key: PrivateKey,
        early_data: bytes | None,
        remote_peer: ID,
        remote_static_key: PublicKey,
    ) -> None:
        self.protocol_name = b"Noise_IK_25519_ChaChaPoly_SHA256"
        self.local_peer = local_peer
        self.libp2p_privkey = libp2p_privkey
        self.noise_static_key = noise_static_key
        self.early_data = early_data
        self.remote_peer = remote_peer
        self.remote_static_key = remote_static_key

    def create_noise_state(self) -> NoiseState:
        noise_state = NoiseState.from_name(self.protocol_name)
        noise_state.set_keypair_from_private_bytes(
            NoiseKeypairEnum.STATIC, self.noise_static_key.to_bytes()
        )
        # Set the remote static key for IK pattern
        noise_state.set_keypair_from_public_bytes(
            NoiseKeypairEnum.REMOTE_STATIC, self.remote_static_key.to_bytes()
        )
        if noise_state.noise_protocol is None:
            raise NoiseStateError("noise_protocol is not initialized")
        return noise_state

    async def handshake_outbound(self, conn: IRawConnection, remote_peer: ID) -> ISecureConn:
        """IK pattern outbound handshake (initiator)."""
        noise_state = self.create_noise_state()
        noise_state.set_as_initiator()
        noise_state.start_handshake()
        
        read_writer = NoiseHandshakeReadWriter(conn, noise_state)
        
        # IK pattern: Send encrypted payload immediately
        our_payload = self.make_handshake_payload()
        msg_1 = our_payload.serialize()
        await read_writer.write_msg(msg_1)
        
        # Receive response
        msg_2 = await read_writer.read_msg()
        peer_handshake_payload = NoiseHandshakePayload.deserialize(msg_2)
        
        # Verify the response
        if not verify_handshake_payload_sig(peer_handshake_payload, self.remote_static_key):
            raise InvalidSignature
        
        remote_peer_id_from_pubkey = ID.from_pubkey(peer_handshake_payload.id_pubkey)
        if remote_peer_id_from_pubkey != remote_peer:
            raise PeerIDMismatchesPubkey(
                f"peer id mismatch: expected={remote_peer}, got={remote_peer_id_from_pubkey}"
            )
        
        if not noise_state.handshake_finished:
            raise HandshakeHasNotFinished("IK handshake not finished")
        
        transport_read_writer = NoiseTransportReadWriter(conn, noise_state)
        return SecureSession(
            local_peer=self.local_peer,
            local_private_key=self.libp2p_privkey,
            remote_peer=remote_peer_id_from_pubkey,
            remote_permanent_pubkey=self.remote_static_key,
            is_initiator=True,
            conn=transport_read_writer,
        )

    async def handshake_inbound(self, conn: IRawConnection) -> ISecureConn:
        """IK pattern inbound handshake (responder)."""
        noise_state = self.create_noise_state()
        noise_state.set_as_responder()
        noise_state.start_handshake()
        
        read_writer = NoiseHandshakeReadWriter(conn, noise_state)
        
        # Receive encrypted payload
        msg_1 = await read_writer.read_msg()
        peer_handshake_payload = NoiseHandshakePayload.deserialize(msg_1)
        
        # Verify the payload
        if not verify_handshake_payload_sig(peer_handshake_payload, self.remote_static_key):
            raise InvalidSignature
        
        # Send response
        our_payload = self.make_handshake_payload()
        msg_2 = our_payload.serialize()
        await read_writer.write_msg(msg_2)
        
        if not noise_state.handshake_finished:
            raise HandshakeHasNotFinished("IK handshake not finished")
        
        transport_read_writer = NoiseTransportReadWriter(conn, noise_state)
        return SecureSession(
            local_peer=self.local_peer,
            local_private_key=self.libp2p_privkey,
            remote_peer=ID.from_pubkey(peer_handshake_payload.id_pubkey),
            remote_permanent_pubkey=self.remote_static_key,
            is_initiator=False,
            conn=transport_read_writer,
        )

Impact Analysis

• Performance: High - IK pattern reduces handshake overhead
• Security: Medium - Maintains security while improving performance
• Complexity: High - Requires pattern implementation and key management
• Storage: Medium - Requires persistent key storage

---

Summary and Recommendations

Priority Order

1. Medium Priority: Static key storage for IK pattern (Issue 2)
   • Performance optimization feature
   • Can be implemented incrementally

Implementation Strategy

1. Phase 3: Add IK pattern implementation
2. Phase 4: Add persistent key storage and advanced features

Testing Requirements

• Performance benchmarking for IK vs XX patterns
• Key cache persistence and expiration testing

Documentation Needs

• Protocol specification compliance notes
• Performance comparison between patterns
• Key management and security considerations
• Interoperability testing results