Day 5: Security Foundations - Encryption, HMAC, Circuit Breaker

• ~5 min read

What I Built

  • Encryption Service: Fernet-based symmetric encryption for sensitive data like IBKR API credentials
  • HMAC Signing: Request authentication with timestamp-based signatures and replay protection
  • Circuit Breaker: External service resilience pattern for IBKR API, database, and market data calls
  • Credentials Management: Encrypted storage and retrieval of service credentials in the database
  • Comprehensive Tests: 25 unit tests covering all security components with edge cases

Code Highlight

# Encryption service using Fernet
from cryptography.fernet import Fernet

class EncryptionService:
    def __init__(self, key: Optional[bytes] = None):
        self.key = key or Fernet.generate_key()
        self.fernet = Fernet(self.key)

    def encrypt(self, plaintext: str) -> str:
        encrypted = self.fernet.encrypt(plaintext.encode())
        return encrypted.decode()

    def decrypt(self, encrypted_text: str) -> str:
        decrypted = self.fernet.decrypt(encrypted_text.encode())
        return decrypted.decode()

# HMAC request signing with replay protection
class HMACSigner:
    def sign_request(self, method: str, url: str, body: Optional[str] = None,
                    headers: Optional[Dict[str, str]] = None,
                    timestamp: Optional[int] = None) -> Dict[str, str]:
        if timestamp is None:
            timestamp = int(time.time())

        message_parts = [method.upper(), url, str(timestamp)]
        if body:
            message_parts.append(body)
        if headers:
            for key, value in sorted(headers.items()):
                message_parts.append(f"{key}:{value}")

        message = "\n".join(message_parts)
        signature = hmac.new(self.secret_key, message.encode(), hashlib.sha256).hexdigest()

        return {
            "X-Timestamp": str(timestamp),
            "X-Signature": signature,
            "X-Algorithm": "sha256"
        }

# Circuit breaker for external service calls
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitBreakerState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None

    async def call(self, func: Callable, *args, **kwargs) -> Any:
        if self.state == CircuitBreakerState.OPEN:
            if self._should_attempt_reset():
                self._half_open_circuit()
            else:
                raise CircuitBreakerOpenException("Circuit breaker is open")

        try:
            result = await asyncio.wait_for(func(*args, **kwargs), timeout=self.timeout)
            self._record_success()
            return result
        except self.expected_exception as e:
            self._record_failure()
            raise e

Architecture Decision

Chose Fernet symmetric encryption over asymmetric crypto for simplicity and performance - the encryption key will be managed via Doppler secrets manager in production. HMAC signing uses SHA-256 with 5-minute replay windows to balance security with usability. Circuit breaker implements the classic three-state pattern (Closed/Open/Half-Open) with configurable thresholds for different service types.

Testing Results

All 25 unit tests pass, covering critical security scenarios:

  • Encryption round-trip with different key instances
  • HMAC signature creation and verification with replay protection
  • Circuit breaker state transitions (closed → open → half-open → closed)
  • Credentials storage, retrieval, update, and deletion
  • Error handling for invalid inputs and expired signatures

Next Steps

Day 6 will focus on credential encryption integration with Doppler, HMAC signing for IBKR API calls, and circuit breakers for external service resilience.

Follow @therealkamba on X for regular updates. View all posts →