Operational Security (OPSEC)

OPSEC is about controlling information flow to prevent adversaries from learning about your operations, capabilities, and vulnerabilities. For agents, this means preventing correlation between activities and protecting operational details.

The OPSEC Process

1. Identify critical information: What must be protected?
2. Analyze threats: Who wants this information?
3. Analyze vulnerabilities: How could information leak?
4. Assess risks: Likelihood × impact
5. Apply countermeasures: Mitigate identified risks

Critical Information for Agents

Information	Risk if Exposed	Adversary Use
IP address	Identity linking	Location, legal action
Transaction patterns	Behavior analysis	Prediction, targeting
Balance amounts	Targeting	Attack prioritization
Key relationships	Wallet clustering	Full exposure
Timing patterns	Traffic analysis	Activity correlation
Communication graphs	Social analysis	Network mapping

Network-Level OPSEC

IP Address Protection

import socks
import socket

def configure_tor_proxy():
    """Route all traffic through Tor."""
    socks.set_default_proxy(socks.SOCKS5, "127.0.0.1", 9050)
    socket.socket = socks.socksocket

# For requests library
proxies = {
    'http': 'socks5h://127.0.0.1:9050',
    'https': 'socks5h://127.0.0.1:9050'
}

async def fetch_with_tor(url: str) -> str:
    """Fetch URL through Tor."""
    async with aiohttp.ClientSession() as session:
        async with session.get(
            url,
            proxy="socks5://127.0.0.1:9050"
        ) as response:
            return await response.text()

Circuit Isolation

Use different Tor circuits for different identities:

class TorCircuitManager:
    """Manage separate Tor circuits per identity."""

    def __init__(self):
        self.circuits = {}

    def get_proxy_for_identity(self, identity: str) -> dict:
        """Get isolated proxy for identity."""
        if identity not in self.circuits:
            # Request new circuit via control port
            port = self._create_new_circuit()
            self.circuits[identity] = port

        return {
            'http': f'socks5h://127.0.0.1:{self.circuits[identity]}',
            'https': f'socks5h://127.0.0.1:{self.circuits[identity]}'
        }

    def _create_new_circuit(self) -> int:
        """Create new Tor circuit on new port."""
        # Implementation uses Tor control protocol
        # Returns port number for new circuit
        pass

DNS Leak Prevention

# Always use socks5h (DNS through proxy)
proxy = "socks5h://127.0.0.1:9050"  # GOOD: h means DNS through Tor
proxy = "socks5://127.0.0.1:9050"   # BAD: DNS leaks to local resolver

# Or resolve through Tor explicitly
async def resolve_through_tor(hostname: str) -> str:
    """Resolve DNS through Tor."""
    # Use Tor's RESOLVE extension
    pass

Transaction OPSEC

Bitcoin Transaction Privacy

Technique	What It Protects	Implementation
New address per tx	Sender privacy	HD wallet
CoinJoin	Transaction graph	Wasabi, JoinMarket
PayJoin	Payment detection	BIP-78
Timing jitter	Traffic analysis	Random delays
Amount decoration	Amount analysis	Add/remove small amounts

import random
import asyncio

async def send_with_timing_jitter(
    destination: str,
    amount_sats: int,
    max_jitter_seconds: int = 300
) -> dict:
    """
    Send payment with random delay to prevent timing correlation.
    """
    # Random delay between 0 and max_jitter
    delay = random.uniform(0, max_jitter_seconds)
    await asyncio.sleep(delay)

    # Now send
    return await send_payment(destination, amount_sats)


def add_amount_jitter(amount_sats: int, jitter_pct: float = 0.01) -> int:
    """
    Add small random variation to amount.

    Prevents correlation by exact amount matching.
    """
    jitter = int(amount_sats * jitter_pct * random.uniform(-1, 1))
    return amount_sats + jitter

Lightning Privacy

Lightning has better base privacy but still requires care:

Concern	Risk	Mitigation
Node ID	Identity linking	Use Tor, separate nodes
Channel graph	Capacity exposure	Private channels
Invoice requests	Payment correlation	Single-use invoices
Routing	Sender/receiver correlation	Use MPP, trampoline

async def create_private_invoice(
    amount_sats: int,
    description: str
) -> dict:
    """Create invoice that doesn't expose our node."""
    return await lightning.create_invoice(
        amount_sats=amount_sats,
        description=description,
        private=True,  # Include route hints but not our pubkey
        expiry=3600    # Short expiry reduces linkability window
    )

Nostr Privacy

Concern	Risk	Mitigation
Pubkey reuse	Activity correlation	Multiple identities
Relay logging	Traffic analysis	Multiple relays, Tor
Metadata	Timing correlation	Jitter, batching
Content	Context correlation	Separate topics per identity

class NostrIdentityManager:
    """Manage multiple Nostr identities."""

    def __init__(self):
        self.identities = {}

    def get_identity_for_purpose(self, purpose: str) -> dict:
        """Get appropriate identity for purpose."""
        if purpose not in self.identities:
            # Generate new identity
            self.identities[purpose] = generate_nostr_keypair()

        return self.identities[purpose]

    async def post_with_identity(
        self,
        purpose: str,
        content: str,
        relays: list[str] | None = None
    ) -> dict:
        """Post from purpose-specific identity."""
        identity = self.get_identity_for_purpose(purpose)

        # Use purpose-specific relays if provided
        target_relays = relays or self._get_relays_for_purpose(purpose)

        return await nostr.post(
            content=content,
            private_key=identity["nsec"],
            relays=target_relays
        )

Identity Separation

The Separation Matrix

┌─────────────────────────────────────────────────────────┐
│                    IDENTITY SEPARATION                   │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌───────────┐    ┌───────────┐    ┌───────────┐       │
│  │ Trading   │    │ Services  │    │ Personal  │       │
│  │ Identity  │    │ Identity  │    │ Identity  │       │
│  └─────┬─────┘    └─────┬─────┘    └─────┬─────┘       │
│        │                │                │             │
│   ┌────┴────┐      ┌────┴────┐      ┌────┴────┐       │
│   │BTC Keys │      │BTC Keys │      │BTC Keys │       │
│   │LN Node  │      │LN Node  │      │LN Node  │       │
│   │Nostr ID │      │Nostr ID │      │Nostr ID │       │
│   │Tor Circ │      │Tor Circ │      │Tor Circ │       │
│   └─────────┘      └─────────┘      └─────────┘       │
│                                                         │
│         NO CROSS-CONTAMINATION BETWEEN IDENTITIES      │
└─────────────────────────────────────────────────────────┘

Implementation

class IdentityIsolation:
    """Strict identity isolation for agent operations."""

    def __init__(self):
        self.identities: dict[str, dict] = {}

    def create_identity(self, name: str) -> dict:
        """Create completely isolated identity."""
        identity = {
            "name": name,
            "bitcoin": {
                "seed": generate_seed_phrase(),
                "addresses": []
            },
            "lightning": {
                "node_seed": os.urandom(32).hex(),
                "macaroon": None  # Set when node created
            },
            "nostr": generate_nostr_keypair(),
            "tor_circuit": self._create_tor_circuit()
        }

        self.identities[name] = identity
        return identity

    def get_session(self, identity_name: str) -> "IsolatedSession":
        """Get session object for isolated operations."""
        identity = self.identities[identity_name]
        return IsolatedSession(identity)


class IsolatedSession:
    """Session bound to single identity."""

    def __init__(self, identity: dict):
        self.identity = identity
        self._proxy = identity["tor_circuit"]["proxy"]

    async def bitcoin_send(self, address: str, amount: int) -> dict:
        """Send Bitcoin using this identity's keys."""
        # Uses identity-specific keys and Tor circuit
        pass

    async def lightning_pay(self, invoice: str) -> dict:
        """Pay via this identity's Lightning node."""
        pass

    async def nostr_post(self, content: str) -> dict:
        """Post from this identity's Nostr key."""
        pass

Operational Hygiene

Time-Based Patterns

class OperationScheduler:
    """Schedule operations to avoid timing patterns."""

    def __init__(self):
        self.operation_history = []

    async def schedule_operation(
        self,
        operation: Callable,
        min_delay: int = 60,
        max_delay: int = 3600
    ):
        """Schedule operation with random delay."""
        # Check recent operations for patterns
        if self._detect_pattern():
            # Add extra jitter if pattern detected
            max_delay *= 2

        delay = random.uniform(min_delay, max_delay)
        await asyncio.sleep(delay)

        result = await operation()

        self.operation_history.append({
            "timestamp": datetime.utcnow(),
            "operation_type": operation.__name__
        })

        return result

    def _detect_pattern(self) -> bool:
        """Detect if recent operations show timing pattern."""
        if len(self.operation_history) < 5:
            return False

        # Check for regular intervals
        intervals = []
        for i in range(1, len(self.operation_history)):
            delta = (
                self.operation_history[i]["timestamp"] -
                self.operation_history[i-1]["timestamp"]
            ).total_seconds()
            intervals.append(delta)

        # If standard deviation is low, pattern detected
        import statistics
        if len(intervals) >= 3:
            std_dev = statistics.stdev(intervals)
            mean = statistics.mean(intervals)
            if std_dev / mean < 0.3:  # Coefficient of variation < 30%
                return True

        return False

Information Compartmentalization

# WRONG: Leaking operational details in logs
logger.info(f"Sending {amount} sats to {address} from wallet {wallet_id}")

# BETTER: Minimal logging
logger.info(f"Payment initiated: {operation_id}")

# Store details in encrypted, separate log
encrypted_log.write({
    "operation_id": operation_id,
    "details": encrypt({
        "amount": amount,
        "address": address,
        "wallet": wallet_id
    })
})

OPSEC Checklist

Before Deployment

• Tor configured and tested
• Identities properly separated
• No correlation between keys
• Logging minimized and encrypted
• Timing jitter implemented

Ongoing Operations

• Monitor for pattern formation
• Rotate identities periodically
• Review logs for leaks
• Test isolation regularly
• Update threat model

---

Machine-Readable Summary

{
  "topic": "opsec",
  "audience": "ai-agents",
  "protection_layers": [
    "network",
    "transaction",
    "identity"
  ],
  "critical_practices": [
    "tor_routing",
    "identity_separation",
    "timing_jitter",
    "dns_leak_prevention"
  ],
  "identities_should_not_share": [
    "ip_addresses",
    "bitcoin_keys",
    "lightning_nodes",
    "nostr_keys",
    "tor_circuits"
  ]
}