HMAC

const crypto = require('crypto');

// ✅ HMAC - Authentification Messages Sécurisée 2024
class HMACManager {
  constructor(options = {}) {
    // Configuration recommandée
    this.config = {
      algorithm: 'sha256',           // HMAC-SHA256 recommandé
      keyLength: 64,                 // 512-bit key (forte sécurité)
      encoding: 'hex',               // Format sortie
      timestampWindow: 300           // 5 minutes fenêtre temporelle
    };
    
    // Clé maître sécurisée (production: HSM/Vault)
    this.masterKey = options.key || this.generateSecureKey();
    
    // Cache signatures pour protection replay
    this.usedSignatures = new Map();
  }

  // Génération clé cryptographiquement forte
  generateSecureKey(length = 64) {
    return crypto.randomBytes(length);
  }

  // HMAC standard avec message
  sign(message, key = null) {
    const signingKey = key || this.masterKey;
    
    const hmac = crypto.createHmac(this.config.algorithm, signingKey);
    hmac.update(message);
    
    return {
      signature: hmac.digest(this.config.encoding),
      algorithm: `HMAC-${this.config.algorithm.toUpperCase()}`,
      message: message,
      timestamp: Date.now()
    };
  }

  // Vérification HMAC avec timing constant
  verify(message, signature) {
    const expected = this.sign(message);
    const expectedSignature = expected.signature;
    
    // Comparaison timing-safe (critique sécurité)
    const isValid = crypto.timingSafeEqual(
      Buffer.from(signature, this.config.encoding),
      Buffer.from(expectedSignature, this.config.encoding)
    );
    
    return {
      valid: isValid,
      algorithm: `HMAC-${this.config.algorithm.toUpperCase()}`,
      timing_safe: true,
      verified_at: new Date()
    };
  }

  // HMAC avec protection replay (timestamp + nonce)
  signWithTimestamp(message) {
    const timestamp = Math.floor(Date.now() / 1000);
    const nonce = crypto.randomBytes(16).toString('hex');
    
    // Construction payload avec métadonnées sécurité
    const payload = `${timestamp}.${nonce}.${message}`;
    const signature = this.sign(payload);
    
    return {
      message: message,
      timestamp: timestamp,
      nonce: nonce,
      signature: signature.signature,
      algorithm: signature.algorithm,
      expires_at: timestamp + this.config.timestampWindow
    };
  }

  // Vérification avec fenêtre temporelle
  verifyWithTimestamp(message, signature, timestamp, nonce) {
    const now = Math.floor(Date.now() / 1000);
    
    // Vérifier expiration
    if (timestamp + this.config.timestampWindow < now) {
      return {
        valid: false,
        error: 'EXPIRED',
        message: 'Signature expired',
        timestamp_diff: now - timestamp
      };
    }
    
    // Vérifier replay (signature déjà utilisée)
    const signatureId = `${signature}-${timestamp}-${nonce}`;
    if (this.usedSignatures.has(signatureId)) {
      return {
        valid: false,
        error: 'REPLAY_ATTACK',
        message: 'Signature already used'
      };
    }
    
    // Reconstituer payload et vérifier
    const payload = `${timestamp}.${nonce}.${message}`;
    const verification = this.verify(payload, signature);
    
    if (verification.valid) {
      // Marquer signature comme utilisée
      this.usedSignatures.set(signatureId, now);
      
      // Nettoyage périodique du cache
      this.cleanupUsedSignatures();
    }
    
    return {
      ...verification,
      replay_protected: true,
      timestamp_valid: true,
      signature_id: signatureId
    };
  }

  // HMAC pour requêtes API (AWS Signature V4 style)
  signApiRequest(method, url, body, headers) {
    const timestamp = Math.floor(Date.now() / 1000);
    const nonce = crypto.randomBytes(8).toString('hex');
    
    // Construction string to sign canonique
    const canonicalHeaders = this.canonicalizeHeaders(headers);
    const stringToSign = [
      method.toUpperCase(),
      url,
      canonicalHeaders,
      timestamp,
      nonce,
      this.hashPayload(body)
    ].join('\n');
    
    const signature = this.sign(stringToSign);
    
    return {
      authorization: `HMAC-SHA256 Credential=api-key,SignedHeaders=${Object.keys(headers).sort().join(';')},Signature=${signature.signature}`,
      'x-timestamp': timestamp,
      'x-nonce': nonce,
      'x-content-hash': this.hashPayload(body),
      signature_details: {
        string_to_sign: stringToSign,
        algorithm: signature.algorithm
      }
    };
  }

  // Vérification requête API
  verifyApiRequest(method, url, body, headers, receivedSignature, timestamp, nonce) {
    // Reconstituer string to sign
    const canonicalHeaders = this.canonicalizeHeaders(headers);
    const stringToSign = [
      method.toUpperCase(),
      url,
      canonicalHeaders,
      timestamp,
      nonce,
      this.hashPayload(body)
    ].join('\n');
    
    return this.verifyWithTimestamp(stringToSign, receivedSignature, parseInt(timestamp), nonce);
  }

  // HMAC pour webhooks (GitHub, Stripe style)
  signWebhook(payload, secret, algorithm = 'sha256') {
    const hmac = crypto.createHmac(algorithm, secret);
    hmac.update(payload);
    
    return {
      signature: `${algorithm}=${hmac.digest('hex')}`,
      raw_signature: hmac.digest('hex'),
      algorithm: algorithm,
      payload_size: payload.length,
      webhook_timestamp: Date.now()
    };
  }

  // Vérification webhook signature
  verifyWebhook(payload, receivedSignature, secret) {
    // Parse signature format "sha256=abc123..."
    const [algorithm, signature] = receivedSignature.split('=');
    
    if (!algorithm || !signature) {
      return { valid: false, error: 'Invalid signature format' };
    }
    
    const expected = this.signWebhook(payload, secret, algorithm);
    
    try {
      const isValid = crypto.timingSafeEqual(
        Buffer.from(signature, 'hex'),
        Buffer.from(expected.raw_signature, 'hex')
      );
      
      return {
        valid: isValid,
        algorithm: `HMAC-${algorithm.toUpperCase()}`,
        webhook_verified: true,
        timing_safe: true
      };
      
    } catch (error) {
      return {
        valid: false,
        error: error.message,
        algorithm: algorithm
      };
    }
  }

  // Utilitaires privées
  canonicalizeHeaders(headers) {
    return Object.keys(headers)
      .sort()
      .map(key => `${key.toLowerCase()}:${headers[key]}`)
      .join('\n');
  }

  hashPayload(payload) {
    return crypto.createHash('sha256')
      .update(payload)
      .digest('hex');
  }

  cleanupUsedSignatures() {
    const now = Math.floor(Date.now() / 1000);
    for (const [key, timestamp] of this.usedSignatures.entries()) {
      if (timestamp + this.config.timestampWindow < now) {
        this.usedSignatures.delete(key);
      }
    }
  }
}

import hmac
import hashlib
import secrets
import time
import json
from typing import Union, Dict, Any, Optional

class HMACManager:
    """Manager HMAC sécurisé pour production Python"""
    
    def __init__(self, key: bytes = None, **kwargs):
        # Configuration recommandée
        self.config = {
            'algorithm': 'sha256',           # HMAC-SHA256 par défaut
            'key_length': 64,                # 512-bit key minimum
            'timestamp_window': 300,         # 5 minutes fenêtre
            'encoding': 'hex'
        }
        
        # Clé maître (production: HSM, KMS, Vault)
        self.master_key = key or self._generate_secure_key()
        
        # Cache signatures utilisées (production: Redis)
        self.used_signatures = {}
    
    def _generate_secure_key(self, length: int = 64) -> bytes:
        """Génération clé cryptographiquement forte"""
        return secrets.token_bytes(length)
    
    def sign(self, message: Union[str, bytes], key: bytes = None) -> Dict[str, Any]:
        """HMAC signature standard"""
        if isinstance(message, str):
            message = message.encode('utf-8')
            
        signing_key = key or self.master_key
        
        signature = hmac.new(
            key=signing_key,
            msg=message,
            digestmod=getattr(hashlib, self.config['algorithm'])
        )
        
        return {
            'signature': signature.hexdigest(),
            'algorithm': f"HMAC-{self.config['algorithm'].upper()}",
            'message_hash': hashlib.sha256(message).hexdigest(),
            'timestamp': int(time.time()),
            'key_length': len(signing_key)
        }
    
    def verify(self, message: Union[str, bytes], signature: str, key: bytes = None) -> Dict[str, Any]:
        """Vérification HMAC avec timing constant"""
        try:
            if isinstance(message, str):
                message = message.encode('utf-8')
                
            signing_key = key or self.master_key
            
            expected_signature = hmac.new(
                key=signing_key,
                msg=message,
                digestmod=getattr(hashlib, self.config['algorithm'])
            ).hexdigest()
            
            # Comparaison timing-safe (critique sécurité)
            is_valid = hmac.compare_digest(signature, expected_signature)
            
            return {
                'valid': is_valid,
                'algorithm': f"HMAC-{self.config['algorithm'].upper()}",
                'timing_safe': True,
                'verified_at': time.time()
            }
            
        except Exception as e:
            return {
                'valid': False,
                'error': str(e),
                'timing_safe': True
            }
    
    def sign_with_timestamp(self, message: str) -> Dict[str, Any]:
        """HMAC avec protection replay"""
        timestamp = int(time.time())
        nonce = secrets.token_hex(16)
        
        # Construction payload avec métadonnées sécurité
        payload = f"{timestamp}.{nonce}.{message}"
        signature_result = self.sign(payload)
        
        return {
            'message': message,
            'timestamp': timestamp,
            'nonce': nonce,
            'signature': signature_result['signature'],
            'algorithm': signature_result['algorithm'],
            'expires_at': timestamp + self.config['timestamp_window']
        }
    
    def verify_with_timestamp(self, message: str, signature: str, timestamp: int, nonce: str) -> Dict[str, Any]:
        """Vérification avec protection replay"""
        current_time = int(time.time())
        
        # Vérifier expiration
        if timestamp + self.config['timestamp_window'] < current_time:
            return {
                'valid': False,
                'error': 'EXPIRED',
                'message': 'Signature expired',
                'timestamp_diff': current_time - timestamp
            }
        
        # Vérifier replay (signature déjà utilisée)
        signature_id = f"{signature}-{timestamp}-{nonce}"
        if signature_id in self.used_signatures:
            return {
                'valid': False,
                'error': 'REPLAY_ATTACK',
                'message': 'Signature already used'
            }
        
        # Reconstituer payload et vérifier
        payload = f"{timestamp}.{nonce}.{message}"
        verification = self.verify(payload, signature)
        
        if verification['valid']:
            # Marquer signature comme utilisée
            self.used_signatures[signature_id] = current_time
            
            # Nettoyage périodique
            self._cleanup_used_signatures()
        
        return {
            **verification,
            'replay_protected': True,
            'timestamp_valid': True,
            'signature_id': signature_id
        }
    
    def sign_api_request(self, method: str, url: str, body: str, headers: Dict[str, str]) -> Dict[str, str]:
        """Signature requête API (style AWS Signature V4)"""
        timestamp = int(time.time())
        nonce = secrets.token_hex(8)
        
        # Construction string to sign canonique
        canonical_headers = self._canonicalize_headers(headers)
        content_hash = hashlib.sha256(body.encode()).hexdigest()
        
        string_to_sign = '\n'.join([
            method.upper(),
            url,
            canonical_headers,
            str(timestamp),
            nonce,
            content_hash
        ])
        
        signature_result = self.sign(string_to_sign)
        
        return {
            'Authorization': f"HMAC-SHA256 Credential=api-key,SignedHeaders={';'.join(sorted(headers.keys()))},Signature={signature_result['signature']}",
            'X-Timestamp': str(timestamp),
            'X-Nonce': nonce,
            'X-Content-Hash': content_hash
        }
    
    def sign_webhook(self, payload: str, secret: str, algorithm: str = 'sha256') -> Dict[str, Any]:
        """Signature webhook (style GitHub, Stripe)"""
        signature = hmac.new(
            key=secret.encode(),
            msg=payload.encode(),
            digestmod=getattr(hashlib, algorithm)
        ).hexdigest()
        
        return {
            'signature': f"{algorithm}={signature}",
            'raw_signature': signature,
            'algorithm': algorithm,
            'payload_size': len(payload),
            'webhook_timestamp': int(time.time())
        }
    
    def verify_webhook(self, payload: str, received_signature: str, secret: str) -> Dict[str, Any]:
        """Vérification signature webhook"""
        try:
            # Parse format "sha256=abc123..."
            if '=' not in received_signature:
                return {'valid': False, 'error': 'Invalid signature format'}
            
            algorithm, signature = received_signature.split('=', 1)
            expected = self.sign_webhook(payload, secret, algorithm)
            
            is_valid = hmac.compare_digest(signature, expected['raw_signature'])
            
            return {
                'valid': is_valid,
                'algorithm': f"HMAC-{algorithm.upper()}",
                'webhook_verified': True,
                'timing_safe': True
            }
            
        except Exception as e:
            return {
                'valid': False,
                'error': str(e),
                'timing_safe': True
            }
    
    def get_security_analysis(self) -> Dict[str, Any]:
        """Analyse configuration sécurité"""
        key_strength = len(self.master_key) * 8
        
        return {
            'algorithm': f"HMAC-{self.config['algorithm'].upper()}",
            'key_strength_bits': key_strength,
            'key_recommendation': 'STRONG' if key_strength >= 256 else 'WEAK',
            'hash_security': self._get_hash_security(self.config['algorithm']),
            'timestamp_window': self.config['timestamp_window'],
            'replay_protection': True,
            'timing_safe_comparison': True,
            'recommendations': self._get_recommendations(key_strength)
        }
    
    def _canonicalize_headers(self, headers: Dict[str, str]) -> str:
        """Canonicalisation headers pour signature"""
        return '\n'.join(
            f"{key.lower()}:{value}"
            for key, value in sorted(headers.items())
        )
    
    def _get_hash_security(self, algorithm: str) -> Dict[str, str]:
        """Évaluation sécurité algorithme hash"""
        security = {
            'sha1': {'level': 'DEPRECATED', 'note': 'Use SHA-256+'},
            'sha256': {'level': 'SECURE', 'note': 'Industry standard'},
            'sha384': {'level': 'SECURE', 'note': 'High security'},
            'sha512': {'level': 'SECURE', 'note': 'Maximum security'}
        }
        return security.get(algorithm, {'level': 'UNKNOWN', 'note': 'Algorithm not recognized'})
    
    def _get_recommendations(self, key_strength: int) -> list:
        """Recommandations basées sur configuration"""
        recommendations = [
            'Use HMAC-SHA256 or stronger',
            'Implement timestamp validation',
            'Enable replay protection',
            'Store keys securely (HSM/KMS)',
            'Rotate keys regularly'
        ]
        
        if key_strength < 256:
            recommendations.insert(0, '⚠️ CRITICAL: Use key ≥256 bits')
        
        return recommendations
    
    def _cleanup_used_signatures(self):
        """Nettoyage cache signatures expirées"""
        current_time = int(time.time())
        expired = [
            sig_id for sig_id, timestamp in self.used_signatures.items()
            if timestamp + self.config['timestamp_window'] < current_time
        ]
        for sig_id in expired:
            del self.used_signatures[sig_id]

# Usage Production
if __name__ == "__main__":
    # Configuration production
    hmac_manager = HMACManager()
    
    # Tests complets
    message = "Données critiques à authentifier"
    
    print("=== HMAC Security Testing ===\n")
    
    # 1. HMAC simple
    signature_result = hmac_manager.sign(message)
    verification = hmac_manager.verify(message, signature_result['signature'])
    print(f"Simple HMAC: {verification}\n")
    
    # 2. HMAC avec protection replay
    timestamped = hmac_manager.sign_with_timestamp(message)
    timestamp_verify = hmac_manager.verify_with_timestamp(
        message, timestamped['signature'], 
        timestamped['timestamp'], timestamped['nonce']
    )
    print(f"Timestamped HMAC: {timestamp_verify}\n")
    
    # 3. Webhook signature
    webhook_payload = json.dumps({'event': 'user.created', 'data': {'id': 123}})
    webhook_secret = 'webhook_secret_key_256_bits_minimum'
    
    webhook_sig = hmac_manager.sign_webhook(webhook_payload, webhook_secret)
    webhook_verify = hmac_manager.verify_webhook(
        webhook_payload, webhook_sig['signature'], webhook_secret
    )
    print(f"Webhook: {webhook_verify}\n")
    
    # 4. Analyse sécurité
    analysis = hmac_manager.get_security_analysis()
    print(f"Security Analysis: {json.dumps(analysis, indent=2)}")